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 07c9f12a50e2..2b937da95a85 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 @@ -976,6 +976,14 @@ public void setKeys(String[] keys) { protected ServiceOffering getServiceOfferingForNodeTypeOnCluster(KubernetesClusterNodeType nodeType, KubernetesCluster cluster) { + return serviceOfferingDao.findById(getServiceOfferingIdForNodeTypeOnCluster(nodeType, cluster)); + } + + /** + * The offering ID set for a node type on the cluster, falling back to the cluster offering. + */ + protected Long getServiceOfferingIdForNodeTypeOnCluster(KubernetesClusterNodeType nodeType, + KubernetesCluster cluster) { Long offeringId = null; Long defaultOfferingId = cluster.getServiceOfferingId(); Long controlOfferingId = cluster.getControlNodeServiceOfferingId(); @@ -994,7 +1002,7 @@ protected ServiceOffering getServiceOfferingForNodeTypeOnCluster(KubernetesClust logger.error(msg); throw new CloudRuntimeException(msg); } - return serviceOfferingDao.findById(offeringId); + return offeringId; } protected boolean isDefaultTemplateUsed() { diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizer.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizer.java new file mode 100644 index 000000000000..d9b5c7812239 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizer.java @@ -0,0 +1,284 @@ +// 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.actionworkers; + +import java.util.HashMap; +import java.util.function.Function; + +import org.apache.cloudstack.utils.bytescale.ByteScaleUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import com.cloud.exception.ManagementServerException; +import com.cloud.exception.ResourceUnavailableException; +import com.cloud.exception.VirtualMachineMigrationException; +import com.cloud.hypervisor.Hypervisor; +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; +import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; +import com.cloud.kubernetes.cluster.utils.GuestResources; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler; +import com.cloud.kubernetes.cluster.utils.NodeAccess; +import com.cloud.kubernetes.cluster.utils.NodeCapacitySnapshot; +import com.cloud.kubernetes.cluster.utils.NodeCapacityReconciliationException; +import com.cloud.offering.ServiceOffering; +import com.cloud.service.dao.ServiceOfferingDao; +import com.cloud.uservm.UserVm; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.vm.UserVmManager; +import com.cloud.vm.UserVmVO; +import com.cloud.vm.dao.UserVmDao; + +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.CONTROL; +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.WORKER; + +/** + * Moves the VMs of a CKS cluster to a new service offering one at a time. For control and worker nodes of a + * running CloudManaged KVM cluster it also makes Kubernetes report the new CPU and memory before the next + * VM is changed. Other VMs, including dedicated etcd VMs, get the plain VM resize. + */ +public class KubernetesClusterNodeOfferingResizer { + protected Logger logger = LogManager.getLogger(getClass()); + + private final KubernetesCluster kubernetesCluster; + private final boolean clusterRunning; + private final UserVmDao userVmDao; + private final ServiceOfferingDao serviceOfferingDao; + private final UserVmManager userVmManager; + private final KubernetesClusterNodeCapacityReconciler reconciler; + private final Function nodeAccessResolver; + + public KubernetesClusterNodeOfferingResizer(KubernetesCluster kubernetesCluster, + boolean clusterRunning, + UserVmDao userVmDao, + ServiceOfferingDao serviceOfferingDao, + UserVmManager userVmManager, + KubernetesClusterNodeCapacityReconciler reconciler, + Function nodeAccessResolver) { + this.kubernetesCluster = kubernetesCluster; + this.clusterRunning = clusterRunning; + this.userVmDao = userVmDao; + this.serviceOfferingDao = serviceOfferingDao; + this.userVmManager = userVmManager; + this.reconciler = reconciler; + this.nodeAccessResolver = nodeAccessResolver; + } + + /** + * Moves one VM to the target offering. A failure is thrown as a CloudRuntimeException that keeps the + * original failure as its cause and says what state the VM and its Kubernetes node were left in. + */ + public void resize(KubernetesClusterVmMapVO vmMap, KubernetesClusterNodeType nodeType, ServiceOffering targetOffering, long deadline) { + UserVmVO vm = userVmDao.findById(vmMap.getVmId()); + if (vm == null) { + throw new CloudRuntimeException(String.format("unable to find the VM with ID %d", vmMap.getVmId())); + } + String context = String.format("unable to scale %s node VM %s (%s) to offering %s", nodeType, vm.getDisplayName(), vm.getUuid(), + targetOffering.getName()); + boolean onTarget = isOnTarget(vm, targetOffering); + String skipReason = getReconciliationSkipReason(vmMap, vm, nodeType); + if (skipReason != null) { + logSkip(vm, skipReason); + if (!onTarget) { + resizeWithoutReconciliation(vm, targetOffering, context); + } + return; + } + if (onTarget) { + finishPendingReconciliation(vm, targetOffering, deadline, context, "it is already on the target offering"); + return; + } + ServiceOffering currentOffering = serviceOfferingDao.findByIdIncludingRemoved(vm.getServiceOfferingId()); + if (currentOffering == null) { + throw new CloudRuntimeException(context + ": the current offering of the VM was not found. The VM was not resized."); + } + if (!KubernetesClusterNodeCapacityReconciler.capacityChanged(currentOffering, targetOffering)) { + resizeWithoutReconciliation(vm, targetOffering, context); + finishPendingReconciliation(vm, targetOffering, deadline, context, "the offering does not change its CPU or RAM"); + return; + } + resizeAndReconcile(vm, targetOffering, deadline, context); + } + + private String getReconciliationSkipReason(KubernetesClusterVmMapVO vmMap, UserVm vm, KubernetesClusterNodeType nodeType) { + if (vmMap.isExternalNode()) { + return "it is an external node"; + } + if (KubernetesCluster.ClusterType.CloudManaged != kubernetesCluster.getClusterType()) { + return "the cluster is not CloudManaged"; + } + if (Hypervisor.HypervisorType.KVM != vm.getHypervisorType()) { + return "it is not a KVM VM"; + } + if (!clusterRunning) { + return "the cluster was not Running"; + } + if (WORKER != nodeType && CONTROL != nodeType) { + return "only control and worker VMs are Kubernetes nodes"; + } + return null; + } + + private void logSkip(UserVm vm, String reason) { + logger.info("Not reconciling the Kubernetes capacity of VM {} in Kubernetes cluster {}: {}", vm, kubernetesCluster, reason); + } + + private boolean isOnTarget(UserVm vm, ServiceOffering targetOffering) { + return targetOffering.getId() == vm.getServiceOfferingId(); + } + + private void resizeWithoutReconciliation(UserVm vm, ServiceOffering targetOffering, String context) { + boolean resized; + try { + resized = userVmManager.upgradeVirtualMachine(vm.getId(), targetOffering.getId(), new HashMap<>()); + } catch (ResourceUnavailableException | ManagementServerException | VirtualMachineMigrationException | RuntimeException e) { + throw new CloudRuntimeException(String.format("%s: %s", context, describe(e)), e); + } + if (!resized) { + throw new CloudRuntimeException(context); + } + } + + /** + * The VM is on the target offering and its CPU and RAM need no change; finish its node only if an earlier + * resize left it unverified. Only the Node is read when there is nothing to finish. + */ + private void finishPendingReconciliation(UserVm vm, ServiceOffering targetOffering, long deadline, String context, String skipReason) { + try { + NodeAccess access = nodeAccessResolver.apply(vm); + NodeCapacitySnapshot node = reconciler.readNode(vm, access); + String clusterUuid = kubernetesCluster.getUuid(); + if (node.hasForeignMarker(clusterUuid)) { + throw new NodeCapacityReconciliationException("the Kubernetes node carries live resize markers of another cluster"); + } + if (!node.isPendingFor(clusterUuid) && !node.hasOwnershipMarker(clusterUuid)) { + logSkip(vm, skipReason); + return; + } + logger.info("Finishing the unverified Kubernetes capacity reconciliation of VM {} in Kubernetes cluster {}", vm, kubernetesCluster); + reconcile(vm, targetOffering, access, node.getUid(), deadline); + } catch (RuntimeException e) { + throw new CloudRuntimeException(String.format("%s: %s. The VM is already on the target offering; retry the same scale operation " + + "to finish reconciling its Kubernetes node.", context, describe(e)), e); + } + } + + private void resizeAndReconcile(UserVmVO vm, ServiceOffering targetOffering, long deadline, String context) { + NodeAccess access; + NodeCapacitySnapshot initial; + try { + access = nodeAccessResolver.apply(vm); + initial = reconciler.readNode(vm, access); + if (initial.hasForeignMarker(kubernetesCluster.getUuid())) { + throw new NodeCapacityReconciliationException("the Kubernetes node carries live resize markers of another cluster"); + } + GuestResources guest = reconciler.readGuest(vm, access); + if (KubernetesClusterNodeCapacityReconciler.isGuestAboveTarget(guest, targetOffering)) { + throw new CloudRuntimeException(String.format("the VM already has %d CPUs and %d MiB of memory online, more than the requested " + + "offering; use an offering at least as large as the VM's current resources", guest.getOnlineCpus(), + guest.getOnlineMemoryBytes() / ByteScaleUtils.MiB)); + } + } catch (RuntimeException e) { + throw new CloudRuntimeException(String.format("%s: %s. The VM was not resized and its Kubernetes node was not changed.", + context, describe(e)), e); + } + NodeCapacitySnapshot started; + try { + started = reconciler.startResize(kubernetesCluster, vm, access, initial.getUid()); + } catch (RuntimeException e) { + throw new CloudRuntimeException(String.format("%s: %s. The VM was not resized; its Kubernetes node may already be marked for " + + "the resize and cordoned. Retry the same scale operation.", context, describe(e)), e); + } + try { + UserVmVO resizedVm = upgrade(vm, targetOffering); + reconcile(resizedVm, targetOffering, access, initial.getUid(), deadline); + } catch (ResourceUnavailableException | ManagementServerException | VirtualMachineMigrationException | RuntimeException e) { + String recovery = recoverAfterFailure(vm, targetOffering, access, initial, started, e); + throw new CloudRuntimeException(String.format("%s: %s. %s", context, describe(e), recovery), e); + } + } + + private UserVmVO upgrade(UserVm vm, ServiceOffering targetOffering) + throws ResourceUnavailableException, ManagementServerException, VirtualMachineMigrationException { + if (!userVmManager.upgradeVirtualMachine(vm.getId(), targetOffering.getId(), new HashMap<>())) { + throw new CloudRuntimeException("the VM could not be scaled"); + } + UserVmVO resizedVm = userVmDao.findById(vm.getId()); + if (resizedVm == null || !isOnTarget(resizedVm, targetOffering)) { + throw new CloudRuntimeException("the VM did not reach the target offering"); + } + return resizedVm; + } + + /** + * After a failed or uncertain resize the pending marker always stays, so a retry finds the node. The CKS + * cordon stays only when the VM is on the target offering; otherwise the next scale resizes the VM again + * (KVM only adds what is missing) and the cordon would just keep the node unschedulable until then. + */ + private String recoverAfterFailure(UserVm vm, ServiceOffering targetOffering, NodeAccess access, + NodeCapacitySnapshot initial, NodeCapacitySnapshot started, Exception failure) { + String retry = " Retry the same scale operation to finish it."; + if (Thread.currentThread().isInterrupted()) { + return "The operation was interrupted and the Kubernetes node was left as it is." + retry; + } + UserVmVO currentVm = null; + try { + currentVm = userVmDao.findById(vm.getId()); + } catch (RuntimeException e) { + failure.addSuppressed(e); + } + if (currentVm == null) { + return "The VM could not be read again, so its Kubernetes node was left as it is." + retry; + } + String clusterUuid = kubernetesCluster.getUuid(); + if (isOnTarget(currentVm, targetOffering)) { + String cordon = started.isCordonOwnedBy(clusterUuid) ? " and cordoned" : ""; + return String.format("The VM was resized, but its Kubernetes node is not reconciled yet and stays marked for reconciliation%s.%s", + cordon, retry); + } + String cordon = ""; + // Release only a cordon this attempt added: the node was schedulable before it started. + if (started.isCordonOwnedBy(clusterUuid) && !initial.isUnschedulable()) { + try { + reconciler.releaseCordon(kubernetesCluster, vm, access, initial.getUid()); + cordon = " CKS removed its temporary cordon from the Kubernetes node."; + } catch (RuntimeException e) { + failure.addSuppressed(e); + cordon = String.format(" CKS could not remove its temporary cordon from the Kubernetes node: %s.", describe(e)); + } + } + return "The VM may have been partly resized on its host." + cordon + retry; + } + + private void reconcile(UserVm vm, ServiceOffering targetOffering, NodeAccess access, String nodeUid, long deadline) { + reconciler.onlineGuestResources(access); + GuestResources guest = reconciler.waitForGuestResources(vm, targetOffering, access, deadline); + if (!reconciler.isKubernetesNodeAlreadyResized(reconciler.readNode(vm, access), guest) || !reconciler.isKubeletActive(access)) { + reconciler.restartKubelet(vm, access, deadline); + reconciler.waitForKubernetesResources(vm, guest, access, deadline); + } + NodeCapacitySnapshot node = reconciler.finishResize(kubernetesCluster, vm, access, nodeUid); + logger.info("Kubernetes node of VM {} in Kubernetes cluster {} reports {}m CPU and {} bytes of memory after the resize and is {}", + vm, kubernetesCluster, node.getCapacity().getCpuMillis(), node.getCapacity().getMemoryBytes(), + node.isUnschedulable() ? "unschedulable" : "schedulable"); + } + + private static String describe(Exception e) { + return StringUtils.defaultIfEmpty(e.getMessage(), e.getClass().getName()); + } +} 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 003a852f3dbe..c484fe550909 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 @@ -19,9 +19,8 @@ import java.io.File; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; +import java.util.EnumMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -29,7 +28,6 @@ import java.util.stream.Collectors; import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; -import com.cloud.service.ServiceOfferingVO; import com.cloud.storage.VMTemplateVO; import org.apache.cloudstack.api.ApiCommandResourceType; import org.apache.cloudstack.api.ApiConstants; @@ -44,21 +42,24 @@ import com.cloud.exception.ManagementServerException; import com.cloud.exception.NetworkRuleConflictException; import com.cloud.exception.ResourceUnavailableException; -import com.cloud.exception.VirtualMachineMigrationException; import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; import com.cloud.kubernetes.cluster.KubernetesClusterService; import com.cloud.kubernetes.cluster.KubernetesClusterVO; import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.kubernetes.cluster.utils.KubernetesClusterUtil; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler; +import com.cloud.kubernetes.cluster.utils.NodeAccess; import com.cloud.network.IpAddress; import com.cloud.network.Network; import com.cloud.network.rules.FirewallRule; +import com.cloud.network.rules.PortForwardingRuleVO; import com.cloud.offering.ServiceOffering; import com.cloud.storage.LaunchPermissionVO; import com.cloud.uservm.UserVm; import com.cloud.utils.Pair; import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.utils.net.NetUtils; import com.cloud.utils.ssh.SshHelper; import com.cloud.vm.UserVmVO; import org.apache.logging.log4j.Level; @@ -70,6 +71,9 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModifierActionWorker { + // Control nodes are scaled first and workers last. + private static final List SCALING_ORDER = List.of(CONTROL, ETCD, WORKER); + private Map serviceOfferingNodeTypeMap; private Long clusterSize; private List nodeIds; @@ -80,6 +84,8 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif private Boolean isAutoscalingEnabled; private long scaleTimeoutTime; + protected KubernetesClusterNodeCapacityReconciler kubernetesClusterNodeCapacityReconciler = new KubernetesClusterNodeCapacityReconciler(); + protected KubernetesClusterScaleWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { super(kubernetesCluster, clusterManager); } @@ -212,7 +218,7 @@ private KubernetesClusterVO updateKubernetesClusterEntryForNodeType(final Long n final ServiceOffering newServiceOffering, final boolean updateNodeOffering, boolean updateClusterOffering) throws CloudRuntimeException { final ServiceOffering serviceOffering = newServiceOffering == null ? - serviceOfferingDao.findById(kubernetesCluster.getServiceOfferingId()) : newServiceOffering; + getOfferingIncludingRemovedForNodeType(nodeType) : newServiceOffering; final Long serviceOfferingId = newServiceOffering == null ? null : serviceOffering.getId(); Pair clusterCountAndCapacity = calculateNewClusterCountAndCapacity(newWorkerSize, nodeType, serviceOffering); @@ -229,32 +235,37 @@ private KubernetesClusterVO updateKubernetesClusterEntryForNodeType(final Long n } protected Pair calculateNewClusterCountAndCapacity(Long newWorkerSize, KubernetesClusterNodeType nodeType, ServiceOffering serviceOffering) { - long cores; - long memory; - long totalClusterSize = newWorkerSize == null ? kubernetesCluster.getTotalNodeCount() : (newWorkerSize + kubernetesCluster.getControlNodeCount() + kubernetesCluster.getEtcdNodeCount()); - - if (nodeType == DEFAULT) { - cores = serviceOffering.getCpu() * totalClusterSize; - memory = serviceOffering.getRamSize() * totalClusterSize; - } else { - long nodeCount = getNodeCountForType(nodeType, kubernetesCluster); - Long existingOfferingId = getExistingOfferingIdForNodeType(nodeType, kubernetesCluster); - if (existingOfferingId == null) { - existingOfferingId = serviceOffering.getId(); - } - ServiceOfferingVO previousOffering = serviceOfferingDao.findById(existingOfferingId); - Pair previousNodesCapacity = calculateNodesCapacity(previousOffering, nodeCount); - if (WORKER == nodeType) { - nodeCount = newWorkerSize == null ? kubernetesCluster.getNodeCount() : newWorkerSize; + long cores = 0; + long memory = 0; + // VM offerings have already changed here. Recompute the total from the + // configured node types instead of subtracting a supposedly old VM offering. + for (KubernetesClusterNodeType type : CLUSTER_NODES_TYPES_LIST) { + long nodeCount = WORKER == type && newWorkerSize != null ? newWorkerSize : getNodeCountForType(type, kubernetesCluster); + if (nodeCount == 0) { + continue; } - Pair newNodesCapacity = calculateNodesCapacity(serviceOffering, nodeCount); - Pair newClusterCapacity = calculateClusterNewCapacity(kubernetesCluster, previousNodesCapacity, newNodesCapacity); - cores = newClusterCapacity.first(); - memory = newClusterCapacity.second(); + ServiceOffering offering = nodeType == DEFAULT || nodeType == type ? serviceOffering : getOfferingIncludingRemovedForNodeType(type); + Pair capacity = calculateNodesCapacity(offering, nodeCount); + cores += capacity.first(); + memory += capacity.second(); } return new Pair<>(cores, memory); } + /** + * The offering the existing nodes of a type run with. It may have been deleted since, which must not + * break the cluster's CPU and memory totals. + */ + private ServiceOffering getOfferingIncludingRemovedForNodeType(KubernetesClusterNodeType nodeType) { + Long offeringId = getServiceOfferingIdForNodeTypeOnCluster(nodeType, kubernetesCluster); + ServiceOffering offering = serviceOfferingDao.findByIdIncludingRemoved(offeringId); + if (offering == null) { + logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster %s failed, unable to find the service offering of its %s nodes", + kubernetesCluster.getName(), nodeType), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + return offering; + } + private long getNodeCountForType(KubernetesClusterNodeType nodeType, KubernetesCluster kubernetesCluster) { if (WORKER == nodeType) { return kubernetesCluster.getNodeCount(); @@ -266,16 +277,6 @@ private long getNodeCountForType(KubernetesClusterNodeType nodeType, KubernetesC return kubernetesCluster.getTotalNodeCount(); } - protected Pair calculateClusterNewCapacity(KubernetesCluster kubernetesCluster, - Pair previousNodeTypeCapacity, - Pair newNodeTypeCapacity) { - long previousCores = kubernetesCluster.getCores(); - long previousMemory = kubernetesCluster.getMemory(); - long newCores = previousCores - previousNodeTypeCapacity.first() + newNodeTypeCapacity.first(); - long newMemory = previousMemory - previousNodeTypeCapacity.second() + newNodeTypeCapacity.second(); - return new Pair<>(newCores, newMemory); - } - protected Pair calculateNodesCapacity(ServiceOffering offering, long nodeCount) { return new Pair<>(offering.getCpu() * nodeCount, offering.getRamSize() * nodeCount); } @@ -338,7 +339,7 @@ private void validateKubernetesClusterScaleSizeParameters(KubernetesClusterNodeT } // Check capacity and transition state final long newVmRequiredCount = clusterSize - originalClusterSize; - final ServiceOffering clusterServiceOffering = serviceOfferingDao.findById(kubernetesCluster.getServiceOfferingId()); + final ServiceOffering clusterServiceOffering = getServiceOfferingForNodeTypeOnCluster(nodeType, kubernetesCluster); if (clusterServiceOffering == null) { logTransitStateToFailedIfNeededAndThrow(Level.WARN, String.format("Scaling failed for Kubernetes cluster : %s, cluster service offering not found", kubernetesCluster.getName())); } @@ -361,7 +362,7 @@ private void validateKubernetesClusterScaleSizeParameters(KubernetesClusterNodeT } } - private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, ServiceOffering serviceOffering, + protected void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, ServiceOffering serviceOffering, boolean updateNodeOffering, boolean updateClusterOffering) throws CloudRuntimeException { validateKubernetesClusterScaleOfferingParameters(); List scalingStates = List.of(KubernetesCluster.State.Scaling, KubernetesCluster.State.ScalingStoppedCluster); @@ -374,26 +375,65 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, } final long size = getNodeCountForType(nodeType, kubernetesCluster); List vmList = kubernetesClusterVmMapDao.listByClusterIdAndVmType(kubernetesCluster.getId(), nodeType); - final long tobeScaledVMCount = Math.min(vmList.size(), size); - for (long i = 0; i < tobeScaledVMCount; i++) { - KubernetesClusterVmMapVO vmMapVO = vmList.get((int) i); - UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); - boolean result = false; - try { - result = userVmManager.upgradeVirtualMachine(userVM.getId(), serviceOffering.getId(), new HashMap()); - } catch (RuntimeException | ResourceUnavailableException | ManagementServerException | VirtualMachineMigrationException e) { - logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster : %s failed, unable to scale cluster VM : %s due to %s", kubernetesCluster.getName(), userVM.getDisplayName(), e.getMessage()), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); - } - if (!result) { - logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster : %s failed, unable to scale cluster VM : %s", kubernetesCluster.getName(), userVM.getDisplayName()),kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); - } + final int tobeScaledVMCount = (int) Math.min(vmList.size(), size); + KubernetesClusterNodeOfferingResizer nodeResizer = createNodeOfferingResizer(); + for (int i = 0; i < tobeScaledVMCount; i++) { if (System.currentTimeMillis() > scaleTimeoutTime) { - logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster : %s failed, scaling action timed out", kubernetesCluster.getName()),kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster : %s failed, scaling action timed out after %d of %d %s VMs were scaled", + kubernetesCluster.getName(), i, tobeScaledVMCount, nodeType), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + try { + nodeResizer.resize(vmList.get(i), nodeType, serviceOffering, scaleTimeoutTime); + } catch (RuntimeException e) { + String reason = StringUtils.defaultIfEmpty(e.getMessage(), e.getClass().getName()); + logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster : %s failed, %s", kubernetesCluster.getName(), reason), + kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); } } kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOffering, updateNodeOffering, updateClusterOffering); } + protected KubernetesClusterNodeOfferingResizer createNodeOfferingResizer() { + return new KubernetesClusterNodeOfferingResizer(kubernetesCluster, + KubernetesCluster.State.Running == originalState, + userVmDao, + serviceOfferingDao, + userVmManager, + kubernetesClusterNodeCapacityReconciler, + this::resolveNodeAccess); + } + + protected NodeAccess resolveNodeAccess(UserVm userVM) { + Pair controlAccess = getKubernetesClusterServerIpSshPort(null); + if (StringUtils.isBlank(controlAccess.first())) { + throw new CloudRuntimeException(String.format("Unable to resolve control-plane SSH for Kubernetes cluster %s", kubernetesCluster.getUuid())); + } + if (manager.isDirectAccess(network)) { + if (StringUtils.isBlank(userVM.getPrivateIpAddress())) { + throw new CloudRuntimeException(String.format("Unable to resolve private SSH address for VM %s", userVM.getUuid())); + } + return new NodeAccess(controlAccess.first(), controlAccess.second(), userVM.getPrivateIpAddress(), DEFAULT_SSH_PORT, + getControlNodeLoginUser(), getManagementServerSshPublicKeyFile()); + } + // The node is reached through its own SSH port-forwarding rule. Use that rule's public IP: on Netris + // VPC tiers it is the NAT IP, not the load balancer IP of the cluster. + PortForwardingRuleVO sshRule = portForwardingRulesDao.listByVm(userVM.getId()).stream() + .filter(rule -> rule.getDestinationPortStart() == DEFAULT_SSH_PORT) + .filter(rule -> NetUtils.TCP_PROTO.equalsIgnoreCase(rule.getProtocol())) + .filter(rule -> rule.getNetworkId() == network.getId()) + .filter(rule -> FirewallRule.State.Revoke != rule.getState()) + .findFirst().orElse(null); + if (sshRule == null) { + throw new CloudRuntimeException(String.format("Unable to resolve SSH port-forwarding rule for VM %s", userVM.getUuid())); + } + IpAddress sshIp = sshRule.getSourceIpAddressId() == null ? null : ipAddressDao.findById(sshRule.getSourceIpAddressId()); + if (sshIp == null) { + throw new CloudRuntimeException(String.format("Unable to resolve the public IP of the SSH port-forwarding rule for VM %s", userVM.getUuid())); + } + return new NodeAccess(controlAccess.first(), controlAccess.second(), sshIp.getAddress().addr(), sshRule.getSourcePortStart(), + getControlNodeLoginUser(), getManagementServerSshPublicKeyFile()); + } + private void removeNodesFromCluster(List vmMaps) throws CloudRuntimeException { for (KubernetesClusterVmMapVO vmMapVO : vmMaps) { UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); @@ -601,15 +641,15 @@ public boolean scaleCluster() throws CloudRuntimeException { final boolean autoscalingChanged = isAutoscalingChanged(); ServiceOffering defaultServiceOffering = serviceOfferingNodeTypeMap.getOrDefault(DEFAULT.name(), null); + Map existingOfferings = getExistingOfferingsOfScaledNodeTypes(scaleClusterDefaultOffering, originalClusterSize); - for (KubernetesClusterNodeType nodeType : Arrays.asList(CONTROL, ETCD, WORKER)) { - boolean isWorkerNode = WORKER == nodeType; - final long newVMRequired = (!isWorkerNode || clusterSize == null) ? 0 : clusterSize - originalClusterSize; - if (!scaleClusterDefaultOffering && !serviceOfferingNodeTypeMap.containsKey(nodeType.name()) && newVMRequired == 0) { + for (KubernetesClusterNodeType nodeType : SCALING_ORDER) { + ServiceOffering existingServiceOffering = existingOfferings.get(nodeType); + if (existingServiceOffering == null) { continue; } - - ServiceOffering existingServiceOffering = getExistingServiceOfferingForNodeType(nodeType, kubernetesCluster); + boolean isWorkerNode = WORKER == nodeType; + final long newVMRequired = (!isWorkerNode || clusterSize == null) ? 0 : clusterSize - originalClusterSize; ServiceOffering scalingServiceOffering = serviceOfferingNodeTypeMap.getOrDefault(nodeType.name(), defaultServiceOffering); boolean isNodeOfferingScalingNeeded = isServiceOfferingScalingNeededForNodeType(existingServiceOffering, scalingServiceOffering); boolean updateNodeOffering = serviceOfferingNodeTypeMap.containsKey(nodeType.name()) || isNodeOfferingScalingNeeded; @@ -643,12 +683,28 @@ public boolean scaleCluster() throws CloudRuntimeException { return true; } + /** + * Looks up the current offering of every node type this request changes before any VM is touched, so a + * missing offering stops the scale before, not after, the VMs of another node type were resized. + */ + private Map getExistingOfferingsOfScaledNodeTypes(boolean scaleClusterDefaultOffering, + long originalClusterSize) { + Map existingOfferings = new EnumMap<>(KubernetesClusterNodeType.class); + for (KubernetesClusterNodeType nodeType : SCALING_ORDER) { + boolean workerCountChanges = WORKER == nodeType && clusterSize != null && clusterSize != originalClusterSize; + if (nodeType == ETCD && kubernetesCluster.getEtcdNodeCount() == 0) { + continue; + } + if (!scaleClusterDefaultOffering && !serviceOfferingNodeTypeMap.containsKey(nodeType.name()) && !workerCountChanges) { + continue; + } + existingOfferings.put(nodeType, getExistingServiceOfferingForNodeType(nodeType, kubernetesCluster)); + } + return existingOfferings; + } + private ServiceOffering getExistingServiceOfferingForNodeType(KubernetesClusterNodeType nodeType, KubernetesCluster kubernetesCluster) { Long existingOfferingId = getExistingOfferingIdForNodeType(nodeType, kubernetesCluster); - if (existingOfferingId == null) { - logAndThrow(Level.ERROR, String.format("The Kubernetes cluster %s does not have a service offering set for node type %s", - kubernetesCluster.getName(), nodeType.name())); - } ServiceOffering existingOffering = serviceOfferingDao.findById(existingOfferingId); if (existingOffering == null) { logAndThrow(Level.ERROR, String.format("Cannot find service offering with ID %s set on the Kubernetes cluster %s node type %s", @@ -674,15 +730,8 @@ protected boolean isServiceOfferingScalingNeededForNodeType(ServiceOffering exis } protected Long getExistingOfferingIdForNodeType(KubernetesClusterNodeType nodeType, KubernetesCluster kubernetesCluster) { - List clusterVms = kubernetesClusterVmMapDao.listByClusterIdAndVmType(kubernetesCluster.getId(), nodeType); - if (CollectionUtils.isEmpty(clusterVms)) { - return kubernetesCluster.getServiceOfferingId(); - } - KubernetesClusterVmMapVO clusterVm = clusterVms.get(0); - UserVmVO clusterUserVm = userVmDao.findById(clusterVm.getVmId()); - if (clusterUserVm == null) { - return kubernetesCluster.getServiceOfferingId(); - } - return clusterUserVm.getServiceOfferingId(); + // Cluster metadata is committed only after every selected VM succeeds. + // A partially resized first VM must not make a retry skip the remaining nodes. + return getServiceOfferingIdForNodeTypeOnCluster(nodeType, kubernetesCluster); } } diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/GuestResources.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/GuestResources.java new file mode 100644 index 000000000000..52cc8a01e6dd --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/GuestResources.java @@ -0,0 +1,63 @@ +// 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.utils; + +import org.apache.cloudstack.utils.bytescale.ByteScaleUtils; + +/** CPU and memory as seen inside a node VM. */ +public final class GuestResources { + private final long onlineCpus; + private final long memoryKiB; + private final long onlineMemoryBytes; + private final long memoryBlockBytes; + + /** + * @param onlineCpus CPUs online in the guest + * @param memoryKiB MemTotal from /proc/meminfo, which kubelet reports as the node's memory capacity + * @param onlineMemoryBytes size of the online memory blocks under /sys/devices/system/memory + * @param memoryBlockBytes size of one of those memory blocks + */ + public GuestResources(long onlineCpus, + long memoryKiB, + long onlineMemoryBytes, + long memoryBlockBytes) { + this.onlineCpus = onlineCpus; + this.memoryKiB = memoryKiB; + this.onlineMemoryBytes = onlineMemoryBytes; + this.memoryBlockBytes = memoryBlockBytes; + } + + public long getOnlineCpus() { + return onlineCpus; + } + + public long getMemoryKiB() { + return memoryKiB; + } + + public long getMemoryBytes() { + return memoryKiB * ByteScaleUtils.KiB; + } + + public long getOnlineMemoryBytes() { + return onlineMemoryBytes; + } + + public long getMemoryBlockBytes() { + return memoryBlockBytes; + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java new file mode 100644 index 000000000000..e38ca3702456 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java @@ -0,0 +1,484 @@ +// 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.utils; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.apache.cloudstack.utils.bytescale.ByteScaleUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.offering.ServiceOffering; +import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; +import com.cloud.utils.ssh.SshHelper; +import com.google.gson.JsonElement; +import com.google.gson.JsonNull; +import com.google.gson.JsonObject; +import com.google.gson.JsonParseException; +import com.google.gson.JsonParser; + +/** + * Brings the guest and Kubernetes views of a CKS node's CPU and memory up to date after a live VM resize, + * using kubectl on the control plane and commands on the node itself. + */ +public class KubernetesClusterNodeCapacityReconciler { + protected Logger logger = LogManager.getLogger(getClass()); + + static final long KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES = 16L * ByteScaleUtils.MiB; + static final int MAX_NODE_UPDATE_ATTEMPTS = 3; + static final long POLL_INTERVAL_MS = 5000L; + static final String GET_NODE_COMMAND = "sudo /opt/bin/kubectl get node %s -o json --request-timeout=20s"; + // CPUs online, MemTotal in KiB, memory block size in hex bytes, and the number of online memory blocks. + static final String GUEST_RESOURCES_COMMAND = "getconf _NPROCESSORS_ONLN; awk '/^MemTotal:/ {print $2}' /proc/meminfo; " + + "cat /sys/devices/system/memory/block_size_bytes; cat /sys/devices/system/memory/memory*/state | grep -cx online"; + static final String RESTART_KUBELET_COMMAND = "sudo systemctl restart kubelet"; + // is-active exits non-zero for every state except active, so print the state instead of failing. + static final String KUBELET_STATE_COMMAND = "systemctl is-active kubelet 2>/dev/null || true"; + static final String ONLINE_HOTPLUGGED_RESOURCES = "sudo sh -c 'for cpu in /sys/devices/system/cpu/cpu[0-9]*; do " + + "[ -e \"$cpu/online\" ] || continue; echo 1 > \"$cpu/online\"; done; " + + "for memory in /sys/devices/system/memory/memory[0-9]*; do " + + "[ -e \"$memory/online\" ] || continue; echo online > \"$memory/online\"; done'"; + private static final int CONNECT_TIMEOUT_MS = 10000; + private static final int COMMAND_TIMEOUT_MS = 30000; + private static final Pattern NODE_NAME = Pattern.compile("[a-z0-9][a-z0-9.-]{0,252}"); + private static final Pattern PATCH_VALUE = Pattern.compile("[A-Za-z0-9-]{1,64}"); + private static final Pattern GUEST_VALUE = Pattern.compile("[1-9][0-9]{0,17}"); + private static final Pattern GUEST_HEX_VALUE = Pattern.compile("[0-9a-fA-F]{1,15}"); + private static final Pattern CPU_QUANTITY = Pattern.compile("([0-9]{1,18})(m)?"); + private static final Pattern MEMORY_QUANTITY = Pattern.compile("([0-9]{1,18})(Ki|Mi|Gi|Ti|K|M|G|T)?"); + + /** Whether the target offering changes the CPU count or the memory of the current one. */ + public static boolean capacityChanged(ServiceOffering currentOffering, ServiceOffering targetOffering) { + return !Objects.equals(currentOffering.getCpu(), targetOffering.getCpu()) + || !Objects.equals(currentOffering.getRamSize(), targetOffering.getRamSize()); + } + + /** Whether the guest has more CPUs online, or at least one memory block more, than the target offering provides. */ + public static boolean isGuestAboveTarget(GuestResources guest, ServiceOffering target) { + long targetMemory = target.getRamSize() * ByteScaleUtils.MiB; + return guest.getOnlineCpus() > target.getCpu() || guest.getOnlineMemoryBytes() - targetMemory >= guest.getMemoryBlockBytes(); + } + + /** + * Whether the guest has exactly the target CPUs online and its online memory blocks add up to the target + * memory. MemTotal is not compared with the offering: it leaves out the memory the kernel keeps for + * itself, which grows with the memory the VM booted with. The memory blocks add up to the RAM of the VM, + * rounded to a whole block, so a memory step smaller than one block (128 MiB on most VMs) is not seen. + */ + static boolean guestMatchesTarget(GuestResources guest, ServiceOffering target) { + long targetMemory = target.getRamSize() * ByteScaleUtils.MiB; + return guest.getOnlineCpus() == target.getCpu() + && Math.abs(guest.getOnlineMemoryBytes() - targetMemory) < guest.getMemoryBlockBytes(); + } + + /** + * Whether the Kubernetes node already reports the guest's current CPUs and memory. Kubelet computes + * allocatable as capacity minus its reservations in the same status update, so a Ready node whose + * capacity matches the guest and whose allocatable lies between zero and its capacity is up to date. + */ + public boolean isKubernetesNodeAlreadyResized(NodeCapacitySnapshot node, GuestResources guest) { + NodeResources capacity = node.getCapacity(); + NodeResources allocatable = node.getAllocatable(); + return node.isReady() + && capacity.getCpuMillis() == guest.getOnlineCpus() * 1000L + && Math.abs(capacity.getMemoryBytes() - guest.getMemoryBytes()) <= KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES + && allocatable.getCpuMillis() >= 0 && allocatable.getCpuMillis() <= capacity.getCpuMillis() + && allocatable.getMemoryBytes() >= 0 && allocatable.getMemoryBytes() <= capacity.getMemoryBytes(); + } + + /** Reads the Kubernetes Node of a VM through the control plane. */ + public NodeCapacitySnapshot readNode(UserVm vm, NodeAccess access) { + String output = executeControl(access, String.format(GET_NODE_COMMAND, nodeName(vm))); + try { + return parseNode(output); + } catch (JsonParseException | IllegalArgumentException | ArithmeticException e) { + throw failure("READ_NODE", vm, "unexpected kubectl output: " + e.getMessage(), e); + } + } + + /** Reads the CPUs online, MemTotal and the online memory blocks inside the VM. */ + public GuestResources readGuest(UserVm vm, NodeAccess access) { + String[] values = executeNode(access, GUEST_RESOURCES_COMMAND).trim().split("\\s+"); + if (values.length != 4 || !GUEST_VALUE.matcher(values[0]).matches() || !GUEST_VALUE.matcher(values[1]).matches() + || !GUEST_HEX_VALUE.matcher(values[2]).matches() || !GUEST_VALUE.matcher(values[3]).matches()) { + throw failure("READ_GUEST", vm, "unexpected guest CPU and memory output"); + } + long memoryBlockBytes = Long.parseLong(values[2], 16); + if (memoryBlockBytes == 0) { + throw failure("READ_GUEST", vm, "unexpected guest memory block size"); + } + try { + return new GuestResources(Long.parseLong(values[0]), + Long.parseLong(values[1]), + Math.multiplyExact(Long.parseLong(values[3]), memoryBlockBytes), + memoryBlockBytes); + } catch (ArithmeticException e) { + throw failure("READ_GUEST", vm, "unexpected guest memory size", e); + } + } + + public void onlineGuestResources(NodeAccess access) { + executeNode(access, ONLINE_HOTPLUGGED_RESOURCES); + } + + public boolean isKubeletActive(NodeAccess access) { + return "active".equals(executeNode(access, KUBELET_STATE_COMMAND).trim()); + } + + /** + * Marks the node as having unverified resize work and, unless it is already cordoned, cordons it and + * records that CKS owns the cordon. An operator's cordon is left as it is. + */ + public NodeCapacitySnapshot startResize(KubernetesCluster cluster, UserVm vm, NodeAccess access, String nodeUid) { + String owner = patchValue(cluster.getUuid(), vm); + return updateNode("START", vm, access, nodeUid, node -> { + if (node.hasForeignMarker(owner)) { + throw failure("START", vm, "the Kubernetes node carries live resize markers of another cluster"); + } + if (node.isPendingFor(owner) && node.isUnschedulable()) { + return null; + } + JsonObject annotations = new JsonObject(); + annotations.addProperty(NodeCapacitySnapshot.PENDING_ANNOTATION, owner); + if (node.isUnschedulable()) { + return nodePatch(vm, node, annotations, null); + } + annotations.addProperty(NodeCapacitySnapshot.CORDON_ANNOTATION, owner); + return nodePatch(vm, node, annotations, true); + }); + } + + /** Removes this cluster's markers and, only if CKS owned the cordon, makes the node schedulable again. */ + public NodeCapacitySnapshot finishResize(KubernetesCluster cluster, UserVm vm, NodeAccess access, String nodeUid) { + String owner = cluster.getUuid(); + return updateNode("FINISH", vm, access, nodeUid, node -> { + boolean pending = node.isPendingFor(owner); + boolean ownershipMarker = node.hasOwnershipMarker(owner); + if (!pending && !ownershipMarker) { + return null; + } + JsonObject annotations = new JsonObject(); + if (pending) { + annotations.add(NodeCapacitySnapshot.PENDING_ANNOTATION, JsonNull.INSTANCE); + } + if (ownershipMarker) { + annotations.add(NodeCapacitySnapshot.CORDON_ANNOTATION, JsonNull.INSTANCE); + } + return nodePatch(vm, node, annotations, node.isCordonOwnedBy(owner) ? Boolean.FALSE : null); + }); + } + + /** Removes a CKS cordon and its ownership marker but keeps the pending marker. */ + public NodeCapacitySnapshot releaseCordon(KubernetesCluster cluster, UserVm vm, NodeAccess access, String nodeUid) { + String owner = cluster.getUuid(); + return updateNode("RELEASE_CORDON", vm, access, nodeUid, node -> { + if (!node.hasOwnershipMarker(owner)) { + return null; + } + JsonObject annotations = new JsonObject(); + annotations.add(NodeCapacitySnapshot.CORDON_ANNOTATION, JsonNull.INSTANCE); + return nodePatch(vm, node, annotations, node.isCordonOwnedBy(owner) ? Boolean.FALSE : null); + }); + } + + /** Waits until the guest has the target resources online; fails at once if it has more than the target. */ + public GuestResources waitForGuestResources(UserVm vm, ServiceOffering target, NodeAccess access, long deadline) { + return poll("GUEST_VERIFY", vm, deadline, "guest CPU or memory did not reach the target offering", () -> { + GuestResources guest = readGuest(vm, access); + if (isGuestAboveTarget(guest, target)) { + throw failure("GUEST_VERIFY", vm, String.format("the guest has %d CPUs and %d MiB of memory online, more than the target offering", + guest.getOnlineCpus(), guest.getOnlineMemoryBytes() / ByteScaleUtils.MiB)); + } + return guestMatchesTarget(guest, target) ? guest : null; + }); + } + + /** Restarts kubelet so it reads the new machine resources, and waits until it is active again. */ + public void restartKubelet(UserVm vm, NodeAccess access, long deadline) { + if (currentTimeMillis() >= deadline) { + throw failure("KUBELET_RESTART", vm, "the scale timeout was reached before kubelet was restarted"); + } + executeNode(access, RESTART_KUBELET_COMMAND); + poll("KUBELET_RESTART", vm, deadline, "kubelet did not become active", () -> isKubeletActive(access) ? Boolean.TRUE : null); + } + + public NodeCapacitySnapshot waitForKubernetesResources(UserVm vm, GuestResources guest, NodeAccess access, long deadline) { + return poll("CAPACITY_VERIFY", vm, deadline, "Kubernetes node capacity did not match the resized guest", () -> { + NodeCapacitySnapshot node = readNode(vm, access); + return isKubernetesNodeAlreadyResized(node, guest) ? node : null; + }); + } + + static NodeCapacitySnapshot parseNode(String output) { + JsonObject node = requireObject(JsonParser.parseString(output), "node"); + JsonObject metadata = requireObject(node.get("metadata"), "metadata"); + JsonObject status = requireObject(node.get("status"), "status"); + JsonElement spec = node.get("spec"); + boolean unschedulable = spec != null && spec.isJsonObject() && isTrue(spec.getAsJsonObject().get("unschedulable")); + return new NodeCapacitySnapshot(requireString(metadata.get("uid"), "metadata.uid"), + requireString(metadata.get("resourceVersion"), "metadata.resourceVersion"), + unschedulable, + readAnnotations(metadata.get("annotations")), + isReady(status.get("conditions")), + readResources(status.get("capacity"), "status.capacity"), + readResources(status.get("allocatable"), "status.allocatable")); + } + + static long parseCpuMillis(String value) { + Matcher matcher = CPU_QUANTITY.matcher(value); + if (!matcher.matches()) { + throw new IllegalArgumentException("unsupported Kubernetes CPU quantity " + value); + } + long number = Long.parseLong(matcher.group(1)); + return matcher.group(2) == null ? Math.multiplyExact(number, 1000L) : number; + } + + static long parseMemoryBytes(String value) { + Matcher matcher = MEMORY_QUANTITY.matcher(value); + if (!matcher.matches()) { + throw new IllegalArgumentException("unsupported Kubernetes memory quantity " + value); + } + return Math.multiplyExact(Long.parseLong(matcher.group(1)), memoryUnitBytes(matcher.group(2))); + } + + private static long memoryUnitBytes(String unit) { + if (unit == null) { + return 1L; + } + switch (unit) { + case "Ki": + return ByteScaleUtils.KiB; + case "Mi": + return ByteScaleUtils.MiB; + case "Gi": + return ByteScaleUtils.GiB; + case "Ti": + return ByteScaleUtils.GiB * ByteScaleUtils.KiB; + case "K": + return 1000L; + case "M": + return 1000L * 1000L; + case "G": + return 1000L * 1000L * 1000L; + case "T": + return 1000L * 1000L * 1000L * 1000L; + default: + throw new IllegalArgumentException("unsupported Kubernetes memory unit " + unit); + } + } + + private static JsonObject requireObject(JsonElement element, String name) { + if (element == null || !element.isJsonObject()) { + throw new IllegalArgumentException("missing " + name); + } + return element.getAsJsonObject(); + } + + private static String requireString(JsonElement element, String name) { + if (!isString(element)) { + throw new IllegalArgumentException("missing " + name); + } + return element.getAsString(); + } + + private static boolean isString(JsonElement element) { + return element != null && element.isJsonPrimitive() && element.getAsJsonPrimitive().isString(); + } + + private static boolean isTrue(JsonElement element) { + return element != null && element.isJsonPrimitive() && element.getAsJsonPrimitive().isBoolean() && element.getAsBoolean(); + } + + private static Map readAnnotations(JsonElement element) { + Map annotations = new HashMap<>(); + if (element == null || !element.isJsonObject()) { + return annotations; + } + for (Map.Entry annotation : element.getAsJsonObject().entrySet()) { + if (isString(annotation.getValue())) { + annotations.put(annotation.getKey(), annotation.getValue().getAsString()); + } + } + return annotations; + } + + private static boolean isReady(JsonElement conditions) { + if (conditions == null || !conditions.isJsonArray()) { + return false; + } + for (JsonElement condition : conditions.getAsJsonArray()) { + if (!condition.isJsonObject()) { + continue; + } + JsonObject item = condition.getAsJsonObject(); + if (isString(item.get("type")) && "Ready".equals(item.get("type").getAsString()) + && isString(item.get("status")) && "True".equals(item.get("status").getAsString())) { + return true; + } + } + return false; + } + + private static NodeResources readResources(JsonElement element, String name) { + JsonObject resources = requireObject(element, name); + return new NodeResources(parseCpuMillis(requireString(resources.get("cpu"), name + ".cpu")), + parseMemoryBytes(requireString(resources.get("memory"), name + ".memory"))); + } + + /** + * Applies the patch that patchFor returns for a fresh read of the node, until patchFor finds nothing + * left to change. Every patch carries the resourceVersion it was based on, so the API server rejects it + * if the node changed in the meantime. After a conflict, a rejected update or a lost reply the next + * read shows what was applied. + */ + private NodeCapacitySnapshot updateNode(String phase, UserVm vm, NodeAccess access, String nodeUid, + Function patchFor) { + String nodeName = nodeName(vm); + for (int attempt = 0; attempt <= MAX_NODE_UPDATE_ATTEMPTS; attempt++) { + NodeCapacitySnapshot node = readNode(vm, access); + if (!node.getUid().equals(nodeUid)) { + throw failure(phase, vm, "the Kubernetes node was replaced during the resize"); + } + String patch = patchFor.apply(node); + if (patch == null) { + return node; + } + if (attempt < MAX_NODE_UPDATE_ATTEMPTS) { + applyPatch(nodeName, patch, access); + } + } + throw failure(phase, vm, String.format("the Kubernetes node could not be updated in %d attempts", MAX_NODE_UPDATE_ATTEMPTS)); + } + + private void applyPatch(String nodeName, String patch, NodeAccess access) { + try { + executeControl(access, KubernetesClusterUtil.getKubernetesNodePatchCommand(nodeName, patch)); + } catch (NodeCapacityReconciliationException e) { + if (Thread.currentThread().isInterrupted()) { + throw e; + } + logger.debug("Updating Kubernetes node {} did not succeed, reading it again: {}", nodeName, e.getMessage()); + } + } + + private String nodePatch(UserVm vm, NodeCapacitySnapshot node, JsonObject annotations, Boolean unschedulable) { + JsonObject metadata = new JsonObject(); + metadata.addProperty("resourceVersion", patchValue(node.getResourceVersion(), vm)); + metadata.add("annotations", annotations); + JsonObject patch = new JsonObject(); + patch.add("metadata", metadata); + if (unschedulable != null) { + JsonObject spec = new JsonObject(); + spec.addProperty("unschedulable", unschedulable); + patch.add("spec", spec); + } + return patch.toString(); + } + + /** Values go into a single-quoted shell argument, so allow only characters that need no quoting. */ + private String patchValue(String value, UserVm vm) { + if (value == null || !PATCH_VALUE.matcher(value).matches()) { + throw failure("NODE_UPDATE", vm, "unexpected value in a Kubernetes node update: " + value); + } + return value; + } + + /** Observes once, then keeps observing while time remains; at most one observation follows the deadline. */ + private T poll(String phase, UserVm vm, long deadline, String timeoutMessage, Supplier observation) { + while (true) { + T result = observation.get(); + if (result != null) { + return result; + } + long remaining = deadline - currentTimeMillis(); + if (remaining <= 0) { + throw failure(phase, vm, timeoutMessage); + } + sleep(Math.min(POLL_INTERVAL_MS, remaining)); + } + } + + protected long currentTimeMillis() { + return System.currentTimeMillis(); + } + + protected void sleep(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new NodeCapacityReconciliationException("CKS live resize polling was interrupted", e); + } + } + + protected String executeControl(NodeAccess access, String command) { + return execute(access.getControlAddress(), access.getControlPort(), access, command); + } + + protected String executeNode(NodeAccess access, String command) { + return execute(access.getNodeAddress(), access.getNodePort(), access, command); + } + + private String execute(String address, int port, NodeAccess access, String command) { + Pair result; + try { + result = SshHelper.sshExecute(address, port, access.getUser(), access.getSshKeyFile(), null, command, + CONNECT_TIMEOUT_MS, CONNECT_TIMEOUT_MS, COMMAND_TIMEOUT_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new NodeCapacityReconciliationException(String.format("Command '%s' on %s was interrupted", command, address), e); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + // SshHelper declares Exception, including connection and authentication failures. + throw new NodeCapacityReconciliationException(String.format("Unable to run command '%s' on %s: %s", command, address, e.getMessage()), e); + } + if (!Boolean.TRUE.equals(result.first())) { + throw new NodeCapacityReconciliationException(String.format("Command '%s' failed on %s: %s", + command, address, StringUtils.abbreviate(StringUtils.trimToEmpty(result.second()), 300))); + } + return StringUtils.defaultString(result.second()); + } + + private String nodeName(UserVm vm) { + String name = KubernetesClusterUtil.getKubernetesNodeName(vm); + if (StringUtils.isBlank(name)) { + throw failure("NODE_NAME", vm, "the VM has no host name"); + } + if (!NODE_NAME.matcher(name).matches()) { + throw failure("NODE_NAME", vm, "the VM host name is not a valid Kubernetes node name"); + } + return name; + } + + private NodeCapacityReconciliationException failure(String phase, UserVm vm, String message) { + return failure(phase, vm, message, null); + } + + private NodeCapacityReconciliationException failure(String phase, UserVm vm, String message, Exception cause) { + return new NodeCapacityReconciliationException(String.format("CKS live resize %s failed for VM %s: %s", phase, vm.getUuid(), message), cause); + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java index 023fc9ec7fad..e04debd592c8 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java @@ -50,6 +50,25 @@ public class KubernetesClusterUtil { public static final String CLUSTER_NODE_READY_COMMAND = "sudo /opt/bin/kubectl get nodes | awk '{if ($1 == \"%s\" && $2 == \"Ready\") print $1}'"; public static final String CLUSTER_NODE_VERSION_COMMAND = "sudo /opt/bin/kubectl get nodes | awk '{if ($1 == \"%s\") print $5}'"; + public static final String CLUSTER_NODE_PATCH_COMMAND = "sudo /opt/bin/kubectl patch node %s --type merge --request-timeout=20s -p '%s'"; + + /** + * The Kubernetes node name of a CKS cluster VM, which is its host name in lowercase. + */ + public static String getKubernetesNodeName(final UserVm userVm) { + return StringUtils.lowerCase(userVm.getHostName()); + } + + /** + * Command that applies a JSON merge patch to a node. Annotations and spec.unschedulable in one patch + * change in a single API update, so a failure can't leave the node with only some of them changed. + */ + public static String getKubernetesNodePatchCommand(final String nodeName, final String mergePatch) { + if (StringUtils.contains(mergePatch, '\'')) { + throw new IllegalArgumentException("A node patch must not contain single quotes"); + } + return String.format(CLUSTER_NODE_PATCH_COMMAND, nodeName, mergePatch); + } public static boolean isKubernetesClusterNodeReady(final KubernetesCluster kubernetesCluster, String ipAddress, int port, String user, File sshKeyFile, String nodeName) throws Exception { @@ -108,10 +127,7 @@ public static boolean uncordonKubernetesClusterNode(final KubernetesCluster kube final String user, final File sshKeyFile, final UserVm userVm, final long timeoutTime, final int waitDuration) { - String hostName = userVm.getHostName(); - if (StringUtils.isNotEmpty(hostName)) { - hostName = hostName.toLowerCase(); - } + String hostName = getKubernetesNodeName(userVm); while (System.currentTimeMillis() < timeoutTime) { Pair result = null; try { diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeAccess.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeAccess.java new file mode 100644 index 000000000000..0f25f408bfed --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeAccess.java @@ -0,0 +1,67 @@ +// 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.utils; + +import java.io.File; + +/** SSH endpoints for a CKS node: the control plane runs kubectl, the node itself runs guest commands. */ +public final class NodeAccess { + private final String controlAddress; + private final int controlPort; + private final String nodeAddress; + private final int nodePort; + private final String user; + private final File sshKeyFile; + + public NodeAccess(String controlAddress, + int controlPort, + String nodeAddress, + int nodePort, + String user, + File sshKeyFile) { + this.controlAddress = controlAddress; + this.controlPort = controlPort; + this.nodeAddress = nodeAddress; + this.nodePort = nodePort; + this.user = user; + this.sshKeyFile = sshKeyFile; + } + + public String getControlAddress() { + return controlAddress; + } + + public int getControlPort() { + return controlPort; + } + + public String getNodeAddress() { + return nodeAddress; + } + + public int getNodePort() { + return nodePort; + } + + public String getUser() { + return user; + } + + public File getSshKeyFile() { + return sshKeyFile; + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacityReconciliationException.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacityReconciliationException.java new file mode 100644 index 000000000000..e9ede51eef6b --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacityReconciliationException.java @@ -0,0 +1,32 @@ +// 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.utils; + +import com.cloud.utils.exception.CloudRuntimeException; + +/** An operational failure while refreshing a node's guest or Kubernetes resources. */ +public class NodeCapacityReconciliationException extends CloudRuntimeException { + private static final long serialVersionUID = 1L; + + public NodeCapacityReconciliationException(String message) { + super(message); + } + + public NodeCapacityReconciliationException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacitySnapshot.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacitySnapshot.java new file mode 100644 index 000000000000..2f4b148299b2 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacitySnapshot.java @@ -0,0 +1,106 @@ +// 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.utils; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.lang3.StringUtils; + +/** The parts of a Kubernetes Node object that a live resize reads. */ +public final class NodeCapacitySnapshot { + /** Set while a node has resize work that has not been verified yet; the value is the cluster UUID. */ + public static final String PENDING_ANNOTATION = "cloudstack.apache.org/cks-live-resize-pending"; + /** Set when CKS, not an operator, cordoned the node for a resize; the value is the cluster UUID. */ + public static final String CORDON_ANNOTATION = "cloudstack.apache.org/cks-live-resize"; + + private final String uid; + private final String resourceVersion; + private final boolean unschedulable; + private final Map annotations; + private final boolean ready; + private final NodeResources capacity; + private final NodeResources allocatable; + + public NodeCapacitySnapshot(String uid, + String resourceVersion, + boolean unschedulable, + Map annotations, + boolean ready, + NodeResources capacity, + NodeResources allocatable) { + this.uid = uid; + this.resourceVersion = resourceVersion; + this.unschedulable = unschedulable; + this.annotations = Collections.unmodifiableMap(new HashMap<>(annotations)); + this.ready = ready; + this.capacity = capacity; + this.allocatable = allocatable; + } + + public String getUid() { + return uid; + } + + public String getResourceVersion() { + return resourceVersion; + } + + public boolean isUnschedulable() { + return unschedulable; + } + + public boolean isReady() { + return ready; + } + + public NodeResources getCapacity() { + return capacity; + } + + public NodeResources getAllocatable() { + return allocatable; + } + + public boolean isPendingFor(String clusterUuid) { + return clusterUuid.equals(annotations.get(PENDING_ANNOTATION)); + } + + /** Whether CKS cordoned the node: it is unschedulable and carries this cluster's ownership marker. */ + public boolean isCordonOwnedBy(String clusterUuid) { + return unschedulable && hasOwnershipMarker(clusterUuid); + } + + /** + * Whether this cluster's ownership marker is set. kubectl uncordon keeps annotations, so the marker can + * be left on a node somebody has uncordoned since; it then no longer means that CKS owns a cordon. + */ + public boolean hasOwnershipMarker(String clusterUuid) { + return clusterUuid.equals(annotations.get(CORDON_ANNOTATION)); + } + + /** Whether a resize marker carries another cluster's UUID; such markers are never changed. */ + public boolean hasForeignMarker(String clusterUuid) { + return isForeign(annotations.get(PENDING_ANNOTATION), clusterUuid) + || isForeign(annotations.get(CORDON_ANNOTATION), clusterUuid); + } + + private static boolean isForeign(String owner, String clusterUuid) { + return StringUtils.isNotEmpty(owner) && !owner.equals(clusterUuid); + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeResources.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeResources.java new file mode 100644 index 000000000000..14e5f39dc6df --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeResources.java @@ -0,0 +1,36 @@ +// 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.utils; + +/** CPU and memory quantities from a Kubernetes Node status, such as its capacity or allocatable. */ +public final class NodeResources { + private final long cpuMillis; + private final long memoryBytes; + + public NodeResources(long cpuMillis, long memoryBytes) { + this.cpuMillis = cpuMillis; + this.memoryBytes = memoryBytes; + } + + public long getCpuMillis() { + return cpuMillis; + } + + public long getMemoryBytes() { + return memoryBytes; + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizerTest.java new file mode 100644 index 000000000000..e160304e96c8 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterNodeOfferingResizerTest.java @@ -0,0 +1,563 @@ +// 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.actionworkers; + +import java.util.HashMap; +import java.util.Map; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.InOrder; +import org.mockito.Mockito; + +import com.cloud.exception.ManagementServerException; +import com.cloud.exception.ResourceUnavailableException; +import com.cloud.exception.VirtualMachineMigrationException; +import com.cloud.hypervisor.Hypervisor.HypervisorType; +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; +import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; +import com.cloud.kubernetes.cluster.utils.GuestResources; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler; +import com.cloud.kubernetes.cluster.utils.NodeAccess; +import com.cloud.kubernetes.cluster.utils.NodeCapacityReconciliationException; +import com.cloud.kubernetes.cluster.utils.NodeCapacitySnapshot; +import com.cloud.kubernetes.cluster.utils.NodeResources; +import com.cloud.service.ServiceOfferingVO; +import com.cloud.service.dao.ServiceOfferingDao; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.vm.UserVmManager; +import com.cloud.vm.UserVmVO; +import com.cloud.vm.dao.UserVmDao; + +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.ETCD; +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.WORKER; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; + +public class KubernetesClusterNodeOfferingResizerTest { + private static final String CLUSTER_UUID = "cluster-uuid"; + private static final long DEADLINE = 600000L; + private static final long MIB = 1024L * 1024L; + + private final KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + private final UserVmDao vmDao = Mockito.mock(UserVmDao.class); + private final ServiceOfferingDao offeringDao = Mockito.mock(ServiceOfferingDao.class); + private final UserVmManager vmManager = Mockito.mock(UserVmManager.class); + private final KubernetesClusterNodeCapacityReconciler reconciler = Mockito.mock(KubernetesClusterNodeCapacityReconciler.class); + private final NodeAccess access = new NodeAccess("control", 22, "node", 2222, "cloud", null); + private final ServiceOfferingVO current = offering(10L, 2, 2048); + private final ServiceOfferingVO target = offering(20L, 4, 4096); + private final UserVmVO vm = vm(10L); + private final UserVmVO resizedVm = vm(20L); + private final KubernetesClusterVmMapVO vmMap = Mockito.mock(KubernetesClusterVmMapVO.class); + private final GuestResources resizedGuest = guest(4, 4026532L, 4096); + private boolean clusterRunning = true; + private int accessResolutions; + + @Before + public void setUp() throws Exception { + Mockito.when(cluster.getUuid()).thenReturn(CLUSTER_UUID); + Mockito.when(cluster.getName()).thenReturn("test-cluster"); + Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.CloudManaged); + Mockito.when(vmMap.getVmId()).thenReturn(1L); + Mockito.when(vmDao.findById(1L)).thenReturn(vm); + Mockito.when(offeringDao.findByIdIncludingRemoved(10L)).thenReturn(current); + Mockito.when(vmManager.upgradeVirtualMachine(eq(1L), eq(20L), any())).thenAnswer(invocation -> { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + return true; + }); + Mockito.when(reconciler.readNode(any(), eq(access))).thenReturn(node(false, null, null)); + Mockito.when(reconciler.readGuest(vm, access)).thenReturn(guest(2, 2029380L, 2048)); + Mockito.when(reconciler.startResize(cluster, vm, access, "uid-1")).thenReturn(node(true, CLUSTER_UUID, CLUSTER_UUID)); + Mockito.when(reconciler.waitForGuestResources(any(), eq(target), eq(access), eq(DEADLINE))).thenReturn(resizedGuest); + Mockito.when(reconciler.isKubeletActive(access)).thenReturn(true); + Mockito.when(reconciler.finishResize(eq(cluster), any(), eq(access), eq("uid-1"))).thenReturn(node(false, null, null)); + } + + @Test + public void testResizesAndReconcilesAWorkerNode() throws Exception { + resize(WORKER); + + InOrder order = Mockito.inOrder(reconciler, vmManager); + order.verify(reconciler).readNode(vm, access); + order.verify(reconciler).readGuest(vm, access); + order.verify(reconciler).startResize(cluster, vm, access, "uid-1"); + order.verify(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + order.verify(reconciler).onlineGuestResources(access); + order.verify(reconciler).waitForGuestResources(resizedVm, target, access, DEADLINE); + order.verify(reconciler).restartKubelet(resizedVm, access, DEADLINE); + order.verify(reconciler).waitForKubernetesResources(resizedVm, resizedGuest, access, DEADLINE); + order.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + Mockito.verify(offeringDao, Mockito.never()).findById(anyLong()); + } + + @Test + public void testUpToDateNodeWithActiveKubeletIsNotRestarted() { + Mockito.when(reconciler.isKubernetesNodeAlreadyResized(any(), eq(resizedGuest))).thenReturn(true); + + resize(WORKER); + + Mockito.verify(reconciler, Mockito.never()).restartKubelet(any(), any(), anyLong()); + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + } + + @Test + public void testInactiveKubeletIsRestartedEvenIfTheNodeLooksUpToDate() { + Mockito.when(reconciler.isKubernetesNodeAlreadyResized(any(), eq(resizedGuest))).thenReturn(true); + Mockito.when(reconciler.isKubeletActive(access)).thenReturn(false); + + resize(WORKER); + + Mockito.verify(reconciler).restartKubelet(resizedVm, access, DEADLINE); + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + } + + @Test + public void testDedicatedEtcdVmGetsThePlainResize() throws Exception { + assertPlainResize(ETCD); + } + + @Test + public void testExternalNodeGetsThePlainResize() throws Exception { + Mockito.when(vmMap.isExternalNode()).thenReturn(true); + assertPlainResize(WORKER); + } + + @Test + public void testExternallyManagedClusterGetsThePlainResize() throws Exception { + Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged); + assertPlainResize(WORKER); + } + + @Test + public void testNonKvmVmGetsThePlainResize() throws Exception { + Mockito.when(vm.getHypervisorType()).thenReturn(HypervisorType.VMware); + assertPlainResize(WORKER); + } + + @Test + public void testClusterThatWasNotRunningGetsThePlainResize() throws Exception { + clusterRunning = false; + assertPlainResize(WORKER); + } + + @Test + public void testOfferingWithTheSameCpuAndRamGetsThePlainResizeAndOneNodeRead() throws Exception { + ServiceOfferingVO sameCapacity = offering(10L, 4, 4096); + Mockito.when(offeringDao.findByIdIncludingRemoved(10L)).thenReturn(sameCapacity); + + resize(WORKER); + + Mockito.verify(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + Mockito.verify(reconciler).readNode(vm, access); + Mockito.verifyNoMoreInteractions(reconciler); + } + + @Test + public void testOfferingWithTheSameCpuAndRamFinishesWorkLeftByAnEarlierResize() throws Exception { + ServiceOfferingVO sameCapacity = offering(10L, 4, 4096); + Mockito.when(offeringDao.findByIdIncludingRemoved(10L)).thenReturn(sameCapacity); + Mockito.when(reconciler.readNode(vm, access)).thenReturn(node(true, CLUSTER_UUID, CLUSTER_UUID)); + + resize(WORKER); + + InOrder order = Mockito.inOrder(vmManager, reconciler); + order.verify(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + order.verify(reconciler).waitForGuestResources(vm, target, access, DEADLINE); + order.verify(reconciler).finishResize(cluster, vm, access, "uid-1"); + Mockito.verify(reconciler, Mockito.never()).startResize(any(), any(), any(), any()); + } + + @Test + public void testPlainResizeFailureKeepsTheCause() throws Exception { + ResourceUnavailableException failure = new ResourceUnavailableException("host unavailable", UserVmVO.class, 1L); + Mockito.doThrow(failure).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(ETCD)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("unable to scale ETCD node VM worker-1")); + } + + @Test + public void testPlainResizeRejectedByTheVmApiFails() throws Exception { + Mockito.doReturn(false).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(ETCD)); + } + + @Test + public void testVmOnTargetWithoutMarkersIsSkippedWithoutGuestAccess() throws Exception { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + + resize(WORKER); + + Mockito.verify(reconciler).readNode(resizedVm, access); + Mockito.verifyNoMoreInteractions(reconciler); + Mockito.verifyNoInteractions(vmManager); + } + + @Test + public void testVmOnTargetWithPendingWorkIsFinishedBeforeAnythingElse() throws Exception { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + Mockito.when(reconciler.readNode(resizedVm, access)).thenReturn(node(true, CLUSTER_UUID, null)); + + resize(WORKER); + + Mockito.verify(reconciler).onlineGuestResources(access); + Mockito.verify(reconciler).waitForGuestResources(resizedVm, target, access, DEADLINE); + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + Mockito.verify(reconciler, Mockito.never()).startResize(any(), any(), any(), any()); + Mockito.verifyNoInteractions(vmManager); + } + + @Test + public void testLeftoverOwnershipMarkerOnASchedulableNodeIsCleanedUp() { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + Mockito.when(reconciler.readNode(resizedVm, access)).thenReturn(node(false, null, CLUSTER_UUID)); + + resize(WORKER); + + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + } + + @Test + public void testOwnershipMarkerWithoutPendingMarkerIsStillFinished() { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + Mockito.when(reconciler.readNode(resizedVm, access)).thenReturn(node(true, null, CLUSTER_UUID)); + + resize(WORKER); + + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + } + + @Test + public void testUnreadableNodeOfAVmOnTargetFailsInsteadOfBeingSkipped() { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + NodeCapacityReconciliationException failure = new NodeCapacityReconciliationException("control plane unreachable"); + Mockito.when(reconciler.readNode(resizedVm, access)).thenThrow(failure); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("retry the same scale operation")); + Mockito.verify(reconciler, Mockito.never()).readGuest(any(), any()); + } + + @Test + public void testMarkersOfAnotherClusterAreNotTreatedAsDone() { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + Mockito.when(reconciler.readNode(resizedVm, access)).thenReturn(node(true, "another-cluster", null)); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Mockito.verify(reconciler, Mockito.never()).finishResize(any(), any(), any(), any()); + } + + @Test + public void testVmOnTargetThatIsNotAKubernetesNodeIsLeftAlone() { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + + resize(ETCD); + + Mockito.verifyNoInteractions(reconciler, vmManager); + } + + @Test + public void testGuestAboveTheTargetIsRejectedBeforeAnyChange() { + Mockito.when(reconciler.readGuest(vm, access)).thenReturn(guest(4, 8152860L, 8192)); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("more than the requested offering")); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("The VM was not resized and its Kubernetes node was not changed")); + Mockito.verify(reconciler, Mockito.never()).startResize(any(), any(), any(), any()); + Mockito.verifyNoInteractions(vmManager); + } + + @Test + public void testGuestReadFailureStopsBeforeAnyChange() { + NodeCapacityReconciliationException failure = new NodeCapacityReconciliationException("SSH unavailable"); + Mockito.when(reconciler.readGuest(vm, access)).thenThrow(failure); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Mockito.verify(reconciler, Mockito.never()).startResize(any(), any(), any(), any()); + Mockito.verifyNoInteractions(vmManager); + } + + @Test + public void testStartFailureStopsBeforeTheVmResize() { + NodeCapacityReconciliationException failure = new NodeCapacityReconciliationException("patch failed"); + Mockito.when(reconciler.startResize(cluster, vm, access, "uid-1")).thenThrow(failure); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("may already be marked")); + Mockito.verifyNoInteractions(vmManager); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + } + + @Test + public void testResourceUnavailableReleasesTheCksCordonAndKeepsPendingWork() throws Exception { + assertCordonReleasedAfter(new ResourceUnavailableException("host unavailable", UserVmVO.class, 1L)); + } + + @Test + public void testManagementFailureReleasesTheCksCordonAndKeepsPendingWork() throws Exception { + assertCordonReleasedAfter(new ManagementServerException("upgrade failed")); + } + + @Test + public void testMigrationFailureReleasesTheCksCordonAndKeepsPendingWork() throws Exception { + assertCordonReleasedAfter(new VirtualMachineMigrationException("migration failed")); + } + + @Test + public void testUnexpectedErrorReleasesTheCksCordonAndKeepsItsCause() throws Exception { + assertCordonReleasedAfter(new NullPointerException("programming error")); + } + + @Test + public void testRejectedVmResizeReleasesTheCksCordon() throws Exception { + Mockito.doReturn(false).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("may have been partly resized")); + Mockito.verify(reconciler).releaseCordon(cluster, vm, access, "uid-1"); + } + + @Test + public void testFailureAfterTheVmReachedTheTargetKeepsPendingWorkAndCordon() throws Exception { + ManagementServerException failure = new ManagementServerException("agent answer lost"); + Mockito.doAnswer(invocation -> { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + throw failure; + }).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("stays marked for reconciliation and cordoned")); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + Mockito.verify(reconciler, Mockito.never()).finishResize(any(), any(), any(), any()); + } + + @Test + public void testReconciliationFailureAfterTheResizeKeepsPendingWorkAndCordon() { + NodeCapacityReconciliationException failure = new NodeCapacityReconciliationException("guest verification failed"); + Mockito.when(reconciler.waitForGuestResources(resizedVm, target, access, DEADLINE)).thenThrow(failure); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("stays marked for reconciliation and cordoned")); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + Mockito.verify(reconciler, Mockito.never()).finishResize(any(), any(), any(), any()); + } + + @Test + public void testOperatorCordonIsNeverReleased() throws Exception { + Mockito.when(reconciler.readNode(vm, access)).thenReturn(node(true, null, null)); + Mockito.when(reconciler.startResize(cluster, vm, access, "uid-1")).thenReturn(node(true, CLUSTER_UUID, null)); + Mockito.doReturn(false).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + } + + @Test + public void testCordonOfAnEarlierAttemptIsKeptAfterAFailedRetry() throws Exception { + Mockito.when(reconciler.readNode(vm, access)).thenReturn(node(true, CLUSTER_UUID, CLUSTER_UUID)); + Mockito.doReturn(false).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + } + + @Test + public void testLeftoverOwnershipMarkerDoesNotKeepThisAttemptsCordon() throws Exception { + Mockito.when(reconciler.readNode(vm, access)).thenReturn(node(false, null, CLUSTER_UUID)); + Mockito.doReturn(false).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Mockito.verify(reconciler).releaseCordon(cluster, vm, access, "uid-1"); + } + + @Test + public void testUnreadableVmAfterAFailureLeavesTheNodeAsItIs() throws Exception { + ManagementServerException failure = new ManagementServerException("upgrade failed"); + CloudRuntimeException databaseFailure = new CloudRuntimeException("database unavailable"); + Mockito.when(vmDao.findById(1L)).thenReturn(vm).thenThrow(databaseFailure); + Mockito.doThrow(failure).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("could not be read again")); + Assert.assertSame(databaseFailure, failure.getSuppressed()[0]); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + } + + @Test + public void testFailedCordonReleaseKeepsTheOriginalFailure() throws Exception { + ManagementServerException failure = new ManagementServerException("upgrade failed"); + NodeCapacityReconciliationException releaseFailure = new NodeCapacityReconciliationException("control plane unreachable"); + Mockito.doThrow(failure).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + Mockito.when(reconciler.releaseCordon(cluster, vm, access, "uid-1")).thenThrow(releaseFailure); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertSame(releaseFailure, failure.getSuppressed()[0]); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("could not remove its temporary cordon")); + } + + @Test + public void testInterruptedResizeLeavesTheNodeAsItIs() throws Exception { + Mockito.doAnswer(invocation -> { + Thread.currentThread().interrupt(); + throw new ManagementServerException("interrupted"); + }).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + try { + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("The operation was interrupted")); + Mockito.verify(reconciler, Mockito.never()).releaseCordon(any(), any(), any(), any()); + } finally { + Thread.interrupted(); + } + } + + @Test + public void testRetryOfTheSameScaleConvergesAndReconciles() throws Exception { + Mockito.doThrow(new ManagementServerException("agent timeout")).doAnswer(invocation -> { + Mockito.when(vmDao.findById(1L)).thenReturn(resizedVm); + return true; + }).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + Mockito.verify(reconciler).releaseCordon(cluster, vm, access, "uid-1"); + Mockito.when(reconciler.readNode(vm, access)).thenReturn(node(false, CLUSTER_UUID, null)); + + resize(WORKER); + + Mockito.verify(vmManager, Mockito.times(2)).upgradeVirtualMachine(eq(1L), eq(20L), any()); + Mockito.verify(reconciler).finishResize(cluster, resizedVm, access, "uid-1"); + } + + @Test + public void testMissingCurrentOfferingFailsBeforeAnyChange() { + Mockito.when(offeringDao.findByIdIncludingRemoved(10L)).thenReturn(null); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("current offering of the VM was not found")); + Mockito.verifyNoInteractions(reconciler, vmManager); + } + + @Test + public void testMissingVmFails() { + Mockito.when(vmDao.findById(1L)).thenReturn(null); + + Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Mockito.verifyNoInteractions(reconciler, vmManager); + } + + private void resize(KubernetesClusterNodeType nodeType) { + new KubernetesClusterNodeOfferingResizer(cluster, + clusterRunning, + vmDao, + offeringDao, + vmManager, + reconciler, + userVm -> { + accessResolutions++; + return access; + }).resize(vmMap, nodeType, target, DEADLINE); + } + + private void assertPlainResize(KubernetesClusterNodeType nodeType) throws Exception { + resize(nodeType); + + Mockito.verify(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + Mockito.verifyNoInteractions(reconciler); + Assert.assertEquals(0, accessResolutions); + } + + private void assertCordonReleasedAfter(Exception failure) throws Exception { + Mockito.doThrow(failure).when(vmManager).upgradeVirtualMachine(eq(1L), eq(20L), any()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, () -> resize(WORKER)); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("may have been partly resized")); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("removed its temporary cordon")); + Mockito.verify(reconciler).releaseCordon(cluster, vm, access, "uid-1"); + Mockito.verify(reconciler, Mockito.never()).finishResize(any(), any(), any(), any()); + } + + private static NodeCapacitySnapshot node(boolean unschedulable, String pendingOwner, String cordonOwner) { + Map annotations = new HashMap<>(); + if (pendingOwner != null) { + annotations.put(NodeCapacitySnapshot.PENDING_ANNOTATION, pendingOwner); + } + if (cordonOwner != null) { + annotations.put(NodeCapacitySnapshot.CORDON_ANNOTATION, cordonOwner); + } + return new NodeCapacitySnapshot("uid-1", + "7", + unschedulable, + annotations, + true, + new NodeResources(4000L, 4L * 1024L * 1024L * 1024L), + new NodeResources(3900L, 3L * 1024L * 1024L * 1024L)); + } + + private static GuestResources guest(long cpus, long memTotalKiB, long onlineMemoryMiB) { + return new GuestResources(cpus, memTotalKiB, onlineMemoryMiB * MIB, 128L * MIB); + } + + private static ServiceOfferingVO offering(long id, int cpu, int memory) { + ServiceOfferingVO offering = Mockito.mock(ServiceOfferingVO.class); + Mockito.when(offering.getId()).thenReturn(id); + Mockito.when(offering.getCpu()).thenReturn(cpu); + Mockito.when(offering.getRamSize()).thenReturn(memory); + Mockito.when(offering.getName()).thenReturn("offering-" + id); + return offering; + } + + private static UserVmVO vm(long offeringId) { + UserVmVO vm = Mockito.mock(UserVmVO.class); + Mockito.when(vm.getId()).thenReturn(1L); + Mockito.when(vm.getUuid()).thenReturn("vm-uuid"); + Mockito.when(vm.getDisplayName()).thenReturn("worker-1"); + Mockito.when(vm.getServiceOfferingId()).thenReturn(offeringId); + Mockito.when(vm.getHypervisorType()).thenReturn(HypervisorType.KVM); + return vm; + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerOfferingTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerOfferingTest.java new file mode 100644 index 000000000000..7f4262f48ba1 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerOfferingTest.java @@ -0,0 +1,253 @@ +// 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.actionworkers; + +import java.util.HashMap; +import java.util.List; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.InOrder; +import org.mockito.Mockito; +import org.springframework.test.util.ReflectionTestUtils; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; +import com.cloud.kubernetes.cluster.KubernetesClusterVO; +import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; +import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; +import com.cloud.kubernetes.cluster.utils.NodeAccess; +import com.cloud.network.Network; +import com.cloud.network.dao.IPAddressDao; +import com.cloud.network.dao.IPAddressVO; +import com.cloud.network.rules.FirewallRule; +import com.cloud.network.rules.PortForwardingRuleVO; +import com.cloud.network.rules.dao.PortForwardingRulesDao; +import com.cloud.service.ServiceOfferingVO; +import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.utils.net.Ip; + +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.WORKER; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; + +public class KubernetesClusterScaleWorkerOfferingTest { + private final KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + private final KubernetesClusterManagerImpl manager = Mockito.mock(KubernetesClusterManagerImpl.class); + private final KubernetesClusterVmMapDao vmMapDao = Mockito.mock(KubernetesClusterVmMapDao.class); + private final KubernetesClusterNodeOfferingResizer resizer = Mockito.mock(KubernetesClusterNodeOfferingResizer.class); + private final PortForwardingRulesDao portForwardingRulesDao = Mockito.mock(PortForwardingRulesDao.class); + private final IPAddressDao ipAddressDao = Mockito.mock(IPAddressDao.class); + private final Network network = Mockito.mock(Network.class); + private final ServiceOfferingVO target = Mockito.mock(ServiceOfferingVO.class); + private final KubernetesClusterVmMapVO firstNode = Mockito.mock(KubernetesClusterVmMapVO.class); + private final KubernetesClusterVmMapVO secondNode = Mockito.mock(KubernetesClusterVmMapVO.class); + private final UserVm vm = Mockito.mock(UserVm.class); + private KubernetesClusterScaleWorker worker; + + @Before + public void setUp() { + Mockito.when(cluster.getId()).thenReturn(100L); + Mockito.when(cluster.getName()).thenReturn("test-cluster"); + Mockito.when(cluster.getUuid()).thenReturn("cluster-uuid"); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(cluster.getNodeCount()).thenReturn(2L); + Mockito.when(cluster.getTotalNodeCount()).thenReturn(2L); + worker = Mockito.spy(new KubernetesClusterScaleWorker(cluster, new HashMap<>(), null, null, null, null, null, manager)); + Mockito.when(cluster.getState()).thenReturn(KubernetesCluster.State.Scaling); + worker.kubernetesClusterVmMapDao = vmMapDao; + worker.portForwardingRulesDao = portForwardingRulesDao; + worker.ipAddressDao = ipAddressDao; + ReflectionTestUtils.setField(worker, "network", network); + ReflectionTestUtils.setField(worker, "scaleTimeoutTime", System.currentTimeMillis() + 60000L); + Mockito.doReturn(resizer).when(worker).createNodeOfferingResizer(); + Mockito.doReturn(true).when(worker).stateTransitTo(anyLong(), any()); + Mockito.doNothing().when(worker).deleteTemplateLaunchPermission(); + Mockito.doReturn(Mockito.mock(KubernetesClusterVO.class)).when(worker).updateKubernetesClusterEntry( + any(), any(), any(), any(), any(), any(), any(), eq(WORKER), anyBoolean(), anyBoolean()); + Mockito.doReturn(new Pair<>("control", 2222)).when(worker).getKubernetesClusterServerIpSshPort(null); + Mockito.doReturn("cloud").when(worker).getControlNodeLoginUser(); + Mockito.doReturn(null).when(worker).getManagementServerSshPublicKeyFile(); + List nodes = List.of(firstNode, secondNode); + Mockito.when(vmMapDao.listByClusterId(100L)).thenReturn(nodes); + Mockito.when(vmMapDao.listByClusterIdAndVmType(100L, WORKER)).thenReturn(nodes); + Mockito.when(target.getId()).thenReturn(20L); + Mockito.when(target.getCpu()).thenReturn(4); + Mockito.when(target.getRamSize()).thenReturn(4096); + Mockito.when(network.getId()).thenReturn(200L); + Mockito.when(vm.getId()).thenReturn(1L); + Mockito.when(vm.getUuid()).thenReturn("vm-uuid"); + Mockito.when(vm.getPrivateIpAddress()).thenReturn("10.1.1.11"); + } + + @Test + public void testNodesAreResizedOneAfterAnotherBeforeTheClusterIsUpdated() { + scale(); + + InOrder order = Mockito.inOrder(resizer, worker); + order.verify(resizer).resize(eq(firstNode), eq(WORKER), eq(target), anyLong()); + order.verify(resizer).resize(eq(secondNode), eq(WORKER), eq(target), anyLong()); + order.verify(worker).updateKubernetesClusterEntry(eq(8L), eq(8192L), any(), eq(20L), any(), any(), any(), eq(WORKER), eq(true), eq(false)); + Mockito.verify(worker, Mockito.never()).stateTransitTo(100L, KubernetesCluster.Event.OperationFailed); + } + + @Test + public void testNodeFailureMarksTheScaleFailedAndStopsBeforeTheNextNode() { + CloudRuntimeException failure = new CloudRuntimeException("unable to scale WORKER node VM worker-1"); + Mockito.doThrow(failure).when(resizer).resize(eq(firstNode), eq(WORKER), eq(target), anyLong()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, this::scale); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("unable to scale WORKER node VM worker-1")); + Mockito.verify(worker).stateTransitTo(100L, KubernetesCluster.Event.OperationFailed); + Mockito.verify(resizer, Mockito.never()).resize(eq(secondNode), any(), any(), anyLong()); + assertClusterNotUpdated(); + } + + @Test + public void testUnexpectedErrorFromANodeAlsoMarksTheScaleFailed() { + NullPointerException failure = new NullPointerException(); + Mockito.doThrow(failure).when(resizer).resize(eq(firstNode), eq(WORKER), eq(target), anyLong()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, this::scale); + + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(error.getMessage(), error.getMessage().contains("NullPointerException")); + Mockito.verify(worker).stateTransitTo(100L, KubernetesCluster.Event.OperationFailed); + assertClusterNotUpdated(); + } + + @Test + public void testExpiredTimeoutStopsBeforeTheNextNode() { + Mockito.doAnswer(invocation -> { + ReflectionTestUtils.setField(worker, "scaleTimeoutTime", System.currentTimeMillis() - 1000L); + return null; + }).when(resizer).resize(eq(firstNode), eq(WORKER), eq(target), anyLong()); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, this::scale); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("timed out after 1 of 2 WORKER VMs were scaled")); + Mockito.verify(resizer, Mockito.never()).resize(eq(secondNode), any(), any(), anyLong()); + Mockito.verify(worker).stateTransitTo(100L, KubernetesCluster.Event.OperationFailed); + assertClusterNotUpdated(); + } + + @Test + public void testTimeoutReachedWhileTheLastNodeFinishesIsNotAFailure() { + Mockito.doAnswer(invocation -> { + ReflectionTestUtils.setField(worker, "scaleTimeoutTime", System.currentTimeMillis() - 1000L); + return null; + }).when(resizer).resize(eq(secondNode), eq(WORKER), eq(target), anyLong()); + + scale(); + + Mockito.verify(worker, Mockito.never()).stateTransitTo(100L, KubernetesCluster.Event.OperationFailed); + Mockito.verify(worker).updateKubernetesClusterEntry(any(), any(), any(), eq(20L), any(), any(), any(), eq(WORKER), eq(true), eq(false)); + } + + @Test + public void testCreatedClusterUpdatesMetadataWithoutTouchingVms() { + ReflectionTestUtils.setField(worker, "originalState", KubernetesCluster.State.Created); + + scale(); + + Mockito.verifyNoInteractions(resizer); + Mockito.verify(worker).updateKubernetesClusterEntry(eq(8L), eq(8192L), any(), eq(20L), any(), any(), any(), eq(WORKER), eq(true), eq(false)); + } + + @Test + public void testDirectAccessUsesThePrivateAddressOfTheNode() { + Mockito.when(manager.isDirectAccess(network)).thenReturn(true); + + NodeAccess access = worker.resolveNodeAccess(vm); + + Assert.assertEquals("control", access.getControlAddress()); + Assert.assertEquals(2222, access.getControlPort()); + Assert.assertEquals("10.1.1.11", access.getNodeAddress()); + Assert.assertEquals(22, access.getNodePort()); + Mockito.verifyNoInteractions(portForwardingRulesDao); + } + + @Test + public void testNodeAccessUsesThePublicIpAndPortOfTheSshRule() { + PortForwardingRuleVO udpRule = sshRule("udp", FirewallRule.State.Active, 4L, 2222); + PortForwardingRuleVO revokedRule = sshRule("tcp", FirewallRule.State.Revoke, 4L, 2222); + PortForwardingRuleVO natRule = sshRule("tcp", FirewallRule.State.Active, 5L, 2223); + Mockito.when(portForwardingRulesDao.listByVm(1L)).thenReturn(List.of(udpRule, revokedRule, natRule)); + IPAddressVO natIp = Mockito.mock(IPAddressVO.class); + Mockito.when(natIp.getAddress()).thenReturn(new Ip("203.0.113.5")); + Mockito.when(ipAddressDao.findById(5L)).thenReturn(natIp); + + NodeAccess access = worker.resolveNodeAccess(vm); + + Assert.assertEquals("control", access.getControlAddress()); + Assert.assertEquals("203.0.113.5", access.getNodeAddress()); + Assert.assertEquals(2223, access.getNodePort()); + // On Netris VPC tiers the cluster's public IP is the load balancer IP, not the NAT IP of the SSH rules. + Mockito.verify(worker, Mockito.never()).getVpcTierKubernetesPublicIp(any()); + Mockito.verify(worker, Mockito.never()).getNetworkSourceNatIp(any()); + } + + @Test + public void testMissingSshRuleFails() { + PortForwardingRuleVO revokedRule = sshRule("tcp", FirewallRule.State.Revoke, 5L, 2223); + Mockito.when(portForwardingRulesDao.listByVm(1L)).thenReturn(List.of(revokedRule)); + + Assert.assertThrows(CloudRuntimeException.class, () -> worker.resolveNodeAccess(vm)); + } + + @Test + public void testMissingIpOfTheSshRuleFails() { + PortForwardingRuleVO ruleWithoutIp = sshRule("tcp", FirewallRule.State.Active, 5L, 2223); + Mockito.when(portForwardingRulesDao.listByVm(1L)).thenReturn(List.of(ruleWithoutIp)); + + Assert.assertThrows(CloudRuntimeException.class, () -> worker.resolveNodeAccess(vm)); + } + + @Test + public void testMissingControlPlaneAddressFails() { + Mockito.doReturn(new Pair<>(null, 2222)).when(worker).getKubernetesClusterServerIpSshPort(null); + + Assert.assertThrows(CloudRuntimeException.class, () -> worker.resolveNodeAccess(vm)); + } + + private void scale() { + worker.scaleKubernetesClusterOffering(WORKER, target, true, false); + } + + private void assertClusterNotUpdated() { + Mockito.verify(worker, Mockito.never()).updateKubernetesClusterEntry( + any(), any(), any(), any(), any(), any(), any(), any(), anyBoolean(), anyBoolean()); + } + + private PortForwardingRuleVO sshRule(String protocol, FirewallRule.State state, long sourceIpId, int sourcePort) { + PortForwardingRuleVO rule = Mockito.mock(PortForwardingRuleVO.class); + Mockito.when(rule.getDestinationPortStart()).thenReturn(22); + Mockito.when(rule.getProtocol()).thenReturn(protocol); + Mockito.when(rule.getNetworkId()).thenReturn(200L); + Mockito.when(rule.getState()).thenReturn(state); + Mockito.when(rule.getSourceIpAddressId()).thenReturn(sourceIpId); + Mockito.when(rule.getSourcePortStart()).thenReturn(sourcePort); + return rule; + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java index c9299bdbaa67..997fb6df8c1a 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorkerTest.java @@ -24,7 +24,7 @@ import com.cloud.service.ServiceOfferingVO; import com.cloud.service.dao.ServiceOfferingDao; import com.cloud.utils.Pair; -import com.cloud.vm.UserVmVO; +import com.cloud.utils.exception.CloudRuntimeException; import com.cloud.vm.dao.UserVmDao; import org.junit.Assert; import org.junit.Before; @@ -40,6 +40,8 @@ import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.CONTROL; import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.DEFAULT; +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.ETCD; +import static com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType.WORKER; @RunWith(MockitoJUnitRunner.class) public class KubernetesClusterScaleWorkerTest { @@ -92,21 +94,13 @@ public void testCalculateNewClusterCountAndCapacityAllNodesScaleSize() { @Test public void testCalculateNewClusterCountAndCapacityNodeTypeScaleControlOffering() { long controlNodes = 2L; - long kubernetesClusterId = 10L; - Mockito.when(kubernetesCluster.getId()).thenReturn(kubernetesClusterId); Mockito.when(kubernetesCluster.getControlNodeCount()).thenReturn(controlNodes); - - ServiceOfferingVO existingOffering = Mockito.mock(ServiceOfferingVO.class); - int existingCores = 2; - int existingMemory = 2048; - Mockito.when(existingOffering.getCpu()).thenReturn(existingCores); - Mockito.when(existingOffering.getRamSize()).thenReturn(existingMemory); + Mockito.when(kubernetesCluster.getNodeCount()).thenReturn(4L); + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(defaultOfferingId); int remainingClusterCpu = 8; int remainingClusterMemory = 12288; - Mockito.when(kubernetesCluster.getCores()).thenReturn(remainingClusterCpu + (controlNodes * existingCores)); - Mockito.when(kubernetesCluster.getMemory()).thenReturn(remainingClusterMemory + (controlNodes * existingMemory)); - - Mockito.when(serviceOfferingDao.findById(1L)).thenReturn(existingOffering); + ServiceOfferingVO workerOffering = offering(2, 3072); + Mockito.when(serviceOfferingDao.findByIdIncludingRemoved(defaultOfferingId)).thenReturn(workerOffering); ServiceOfferingVO newOffering = Mockito.mock(ServiceOfferingVO.class); int newCores = 4; @@ -114,12 +108,6 @@ public void testCalculateNewClusterCountAndCapacityNodeTypeScaleControlOffering( Mockito.when(newOffering.getCpu()).thenReturn(newCores); Mockito.when(newOffering.getRamSize()).thenReturn(newMemory); - KubernetesClusterVmMapVO controlNodeVM1 = Mockito.mock(KubernetesClusterVmMapVO.class); - Mockito.when(controlNodeVM1.getVmId()).thenReturn(10L); - UserVmVO userVmVO = Mockito.mock(UserVmVO.class); - Mockito.when(userVmVO.getServiceOfferingId()).thenReturn(defaultOfferingId); - Mockito.when(userVmDao.findById(10L)).thenReturn(userVmVO); - Mockito.when(kubernetesClusterVmMapDao.listByClusterIdAndVmType(kubernetesClusterId, CONTROL)).thenReturn(List.of(controlNodeVM1)); Pair newClusterCapacity = worker.calculateNewClusterCountAndCapacity(null, CONTROL, newOffering); long expectedCores = remainingClusterCpu + (controlNodes * newCores); @@ -128,6 +116,121 @@ public void testCalculateNewClusterCountAndCapacityNodeTypeScaleControlOffering( Assert.assertEquals(expectedMemory, newClusterCapacity.second().longValue()); } + @Test + public void testWorkerResizeRecomputesStaleTotalsWithoutReadingResizedVmOfferings() { + Mockito.when(kubernetesCluster.getControlNodeCount()).thenReturn(1L); + Mockito.when(kubernetesCluster.getNodeCount()).thenReturn(1L); + Mockito.when(kubernetesCluster.getControlNodeServiceOfferingId()).thenReturn(2L); + Mockito.doReturn(offering(3, 3072)).when(serviceOfferingDao).findByIdIncludingRemoved(2L); + + Pair capacity = worker.calculateNewClusterCountAndCapacity(null, WORKER, offering(3, 3072)); + + Assert.assertEquals(Long.valueOf(6), capacity.first()); + Assert.assertEquals(Long.valueOf(6144), capacity.second()); + Mockito.verify(kubernetesCluster, Mockito.never()).getCores(); + Mockito.verify(kubernetesCluster, Mockito.never()).getMemory(); + Mockito.verifyNoInteractions(userVmDao, kubernetesClusterVmMapDao); + } + + @Test + public void testWorkerCountChangePreservesMixedControlAndEtcdOfferings() { + Mockito.when(kubernetesCluster.getControlNodeCount()).thenReturn(1L); + Mockito.when(kubernetesCluster.getEtcdNodeCount()).thenReturn(3L); + Mockito.when(kubernetesCluster.getControlNodeServiceOfferingId()).thenReturn(2L); + Mockito.when(kubernetesCluster.getEtcdNodeServiceOfferingId()).thenReturn(3L); + Mockito.doReturn(offering(4, 4096)).when(serviceOfferingDao).findByIdIncludingRemoved(2L); + Mockito.doReturn(offering(2, 2048)).when(serviceOfferingDao).findByIdIncludingRemoved(3L); + + Pair capacity = worker.calculateNewClusterCountAndCapacity(2L, WORKER, offering(3, 3072)); + + Assert.assertEquals(Long.valueOf(16), capacity.first()); + Assert.assertEquals(Long.valueOf(16384), capacity.second()); + } + + @Test + public void testCapacityFallsBackToClusterOfferingForLegacyNodeTypes() { + Mockito.when(kubernetesCluster.getControlNodeCount()).thenReturn(1L); + Mockito.when(kubernetesCluster.getControlNodeServiceOfferingId()).thenReturn(null); + Mockito.when(kubernetesCluster.getServiceOfferingId()).thenReturn(defaultOfferingId); + Mockito.doReturn(offering(2, 2048)).when(serviceOfferingDao).findByIdIncludingRemoved(defaultOfferingId); + + Pair capacity = worker.calculateNewClusterCountAndCapacity(1L, WORKER, offering(3, 3072)); + + Assert.assertEquals(Long.valueOf(5), capacity.first()); + Assert.assertEquals(Long.valueOf(5120), capacity.second()); + } + + private ServiceOfferingVO offering(int cpu, int memory) { + ServiceOfferingVO offering = Mockito.mock(ServiceOfferingVO.class); + Mockito.when(offering.getCpu()).thenReturn(cpu); + Mockito.when(offering.getRamSize()).thenReturn(memory); + return offering; + } + + @Test + public void testExistingOfferingUsesCommittedMetadataRatherThanPartiallyResizedFirstVm() { + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(17L); + Assert.assertEquals(Long.valueOf(17), worker.getExistingOfferingIdForNodeType(WORKER, kubernetesCluster)); + Mockito.verifyNoInteractions(userVmDao, kubernetesClusterVmMapDao); + } + + @Test + public void testWorkerPlanningUsesNodeOfferingWithoutLegacyGlobalOffering() { + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(17L); + ServiceOfferingVO workerOffering = offering(3, 3072); + Mockito.when(serviceOfferingDao.findById(17L)).thenReturn(workerOffering); + Assert.assertSame(workerOffering, worker.getServiceOfferingForNodeTypeOnCluster(WORKER, kubernetesCluster)); + } + + @Test + public void testGlobalOfferingChangeSkipsAbsentDedicatedEtcdNodes() { + Mockito.when(kubernetesCluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(kubernetesCluster.getServiceOfferingId()).thenReturn(null); + Mockito.when(kubernetesCluster.getControlNodeServiceOfferingId()).thenReturn(11L); + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(12L); + ServiceOfferingVO controlOffering = Mockito.mock(ServiceOfferingVO.class); + ServiceOfferingVO workerOffering = Mockito.mock(ServiceOfferingVO.class); + ServiceOfferingVO target = Mockito.mock(ServiceOfferingVO.class); + Mockito.when(controlOffering.getId()).thenReturn(11L); + Mockito.when(workerOffering.getId()).thenReturn(12L); + Mockito.when(target.getId()).thenReturn(13L); + Mockito.when(serviceOfferingDao.findById(11L)).thenReturn(controlOffering); + Mockito.when(serviceOfferingDao.findById(12L)).thenReturn(workerOffering); + KubernetesClusterScaleWorker scaleWorker = Mockito.spy(new KubernetesClusterScaleWorker(kubernetesCluster, + java.util.Map.of(DEFAULT.name(), target), null, null, null, null, null, clusterManager)); + scaleWorker.serviceOfferingDao = serviceOfferingDao; + Mockito.doNothing().when(scaleWorker).init(); + Mockito.doNothing().when(scaleWorker).scaleKubernetesClusterOffering(Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + Mockito.doReturn(true).when(scaleWorker).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + Assert.assertTrue(scaleWorker.scaleCluster()); + + Mockito.verify(scaleWorker).scaleKubernetesClusterOffering(CONTROL, target, true, false); + Mockito.verify(scaleWorker).scaleKubernetesClusterOffering(WORKER, target, true, true); + Mockito.verify(scaleWorker, Mockito.never()).scaleKubernetesClusterOffering(Mockito.eq(ETCD), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + } + + @Test + public void testRetryUsesUncommittedWorkerOfferingToEnterResizeLoop() { + Mockito.when(kubernetesCluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(11L); + ServiceOfferingVO old = Mockito.mock(ServiceOfferingVO.class); + ServiceOfferingVO target = Mockito.mock(ServiceOfferingVO.class); + Mockito.when(old.getId()).thenReturn(11L); + Mockito.when(target.getId()).thenReturn(12L); + Mockito.when(serviceOfferingDao.findById(11L)).thenReturn(old); + KubernetesClusterScaleWorker scaleWorker = Mockito.spy(new KubernetesClusterScaleWorker(kubernetesCluster, + java.util.Map.of(WORKER.name(), target), null, null, null, null, null, clusterManager)); + scaleWorker.serviceOfferingDao = serviceOfferingDao; + Mockito.doNothing().when(scaleWorker).init(); + Mockito.doNothing().when(scaleWorker).scaleKubernetesClusterOffering(Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + Mockito.doReturn(true).when(scaleWorker).stateTransitTo(Mockito.anyLong(), Mockito.any()); + + Assert.assertTrue(scaleWorker.scaleCluster()); + + Mockito.verify(scaleWorker).scaleKubernetesClusterOffering(WORKER, target, true, false); + } + @Test public void testGetWorkerNodesToRemoveForDownsize_singleRemoval() { @@ -187,4 +290,29 @@ public void testGetWorkerNodesToRemoveForDownsize_noRemoval() { Assert.assertTrue(toRemove.isEmpty()); } + + @Test + public void testUsesLiveResizeReconcilerWithoutComponentContextRegistration() { + Assert.assertNotNull(worker.kubernetesClusterNodeCapacityReconciler); + } + + @Test + public void testMissingOfferingOfALaterNodeTypeFailsBeforeAnyVmIsResized() { + Mockito.when(kubernetesCluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(kubernetesCluster.getControlNodeServiceOfferingId()).thenReturn(11L); + Mockito.when(kubernetesCluster.getWorkerNodeServiceOfferingId()).thenReturn(12L); + ServiceOfferingVO controlOffering = Mockito.mock(ServiceOfferingVO.class); + Mockito.when(serviceOfferingDao.findById(11L)).thenReturn(controlOffering); + ServiceOfferingVO target = Mockito.mock(ServiceOfferingVO.class); + KubernetesClusterScaleWorker scaleWorker = Mockito.spy(new KubernetesClusterScaleWorker(kubernetesCluster, + java.util.Map.of(CONTROL.name(), target, WORKER.name(), target), null, null, null, null, null, clusterManager)); + scaleWorker.serviceOfferingDao = serviceOfferingDao; + Mockito.doNothing().when(scaleWorker).init(); + + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, scaleWorker::scaleCluster); + + Assert.assertTrue(error.getMessage(), error.getMessage().contains("ID 12") && error.getMessage().contains("WORKER")); + + Mockito.verify(scaleWorker, Mockito.never()).scaleKubernetesClusterOffering(Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean()); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerTest.java new file mode 100644 index 000000000000..e3ad8ca710d7 --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerTest.java @@ -0,0 +1,781 @@ +// 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.utils; + +import java.io.File; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.function.ThrowingRunnable; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.offering.ServiceOffering; +import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; +import com.cloud.utils.ssh.SshHelper; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; + +public class KubernetesClusterNodeCapacityReconcilerTest { + private static final String CLUSTER_UUID = "cluster-uuid"; + private static final String PENDING = NodeCapacitySnapshot.PENDING_ANNOTATION; + private static final String CORDON = NodeCapacitySnapshot.CORDON_ANNOTATION; + private static final long MIB = 1024L * 1024L; + private static final long GIB = 1024L * MIB; + + private final KubernetesCluster cluster = cluster(); + private final UserVm vm = vm("worker-1"); + private final NodeAccess access = new NodeAccess("control", 22, "node", 22, "root", null); + private final KubernetesClusterNodeCapacityReconciler reconciler = new KubernetesClusterNodeCapacityReconciler(); + + @Test + public void testParseCpuAndMemoryQuantities() { + Assert.assertEquals(3900L, KubernetesClusterNodeCapacityReconciler.parseCpuMillis("3900m")); + Assert.assertEquals(4000L, KubernetesClusterNodeCapacityReconciler.parseCpuMillis("4")); + Assert.assertEquals(123L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("123")); + Assert.assertEquals(4037048L * 1024L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("4037048Ki")); + Assert.assertEquals(2L * 1024L * 1024L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2Mi")); + Assert.assertEquals(4L * GIB, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("4Gi")); + Assert.assertEquals(2L * 1024L * GIB, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2Ti")); + Assert.assertEquals(2000L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2K")); + Assert.assertEquals(2000000L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2M")); + Assert.assertEquals(2000000000L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2G")); + Assert.assertEquals(2000000000000L, KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("2T")); + } + + @Test + public void testQuantityParsingRejectsMalformedAndOverflowingValues() { + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("4Gi; reboot")); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("1.5Gi")); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("-1")); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseCpuMillis("1e3")); + Assert.assertThrows(ArithmeticException.class, () -> KubernetesClusterNodeCapacityReconciler.parseMemoryBytes("99999999999999999Ti")); + Assert.assertThrows(ArithmeticException.class, () -> KubernetesClusterNodeCapacityReconciler.parseCpuMillis("99999999999999999")); + } + + @Test + public void testParseNodeReadsIdentityMarkersSchedulingAndResources() { + FakeNode node = new FakeNode(); + node.unschedulable = true; + node.annotations.put(PENDING, CLUSTER_UUID); + node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot snapshot = KubernetesClusterNodeCapacityReconciler.parseNode(node.toJson()); + + Assert.assertEquals("uid-1", snapshot.getUid()); + Assert.assertEquals("7", snapshot.getResourceVersion()); + Assert.assertTrue(snapshot.isUnschedulable()); + Assert.assertTrue(snapshot.isReady()); + Assert.assertTrue(snapshot.isPendingFor(CLUSTER_UUID)); + Assert.assertTrue(snapshot.isCordonOwnedBy(CLUSTER_UUID)); + Assert.assertFalse(snapshot.hasForeignMarker(CLUSTER_UUID)); + Assert.assertTrue(snapshot.hasForeignMarker("another-cluster")); + Assert.assertEquals(2000L, snapshot.getCapacity().getCpuMillis()); + Assert.assertEquals(2L * GIB, snapshot.getCapacity().getMemoryBytes()); + Assert.assertEquals(1900L, snapshot.getAllocatable().getCpuMillis()); + Assert.assertEquals(1900L * 1024L * 1024L, snapshot.getAllocatable().getMemoryBytes()); + } + + @Test + public void testParseNodeAcceptsMissingOptionalFields() { + JsonObject node = JsonParser.parseString(new FakeNode().toJson()).getAsJsonObject(); + node.remove("spec"); + node.getAsJsonObject("metadata").remove("annotations"); + node.getAsJsonObject("status").remove("conditions"); + + NodeCapacitySnapshot snapshot = KubernetesClusterNodeCapacityReconciler.parseNode(node.toString()); + + Assert.assertFalse(snapshot.isUnschedulable()); + Assert.assertFalse(snapshot.isReady()); + Assert.assertFalse(snapshot.isPendingFor(CLUSTER_UUID)); + Assert.assertFalse(snapshot.hasForeignMarker(CLUSTER_UUID)); + } + + @Test + public void testParseNodeRejectsMissingRequiredFields() { + JsonObject withoutUid = JsonParser.parseString(new FakeNode().toJson()).getAsJsonObject(); + withoutUid.getAsJsonObject("metadata").remove("uid"); + JsonObject withoutMemory = JsonParser.parseString(new FakeNode().toJson()).getAsJsonObject(); + withoutMemory.getAsJsonObject("status").getAsJsonObject("capacity").remove("memory"); + + String withoutUidJson = withoutUid.toString(); + String withoutMemoryJson = withoutMemory.toString(); + + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseNode(withoutUidJson)); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseNode(withoutMemoryJson)); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconciler.parseNode("[]")); + } + + @Test + public void testMalformedKubectlOutputFailsWithPhaseAndVm() { + FakeReconciler fake = new FakeReconciler(); + fake.nodeOutput = "Error from server: not json"; + + assertPhaseFailure("READ_NODE", () -> fake.readNode(vm, access)); + } + + @Test + public void testNodeNameIsLowercaseInKubectlCommand() { + FakeReconciler fake = new FakeReconciler(); + + fake.readNode(vm("Worker-1"), access); + + Assert.assertEquals("sudo /opt/bin/kubectl get node worker-1 -o json --request-timeout=20s", fake.controlCommands.get(0)); + } + + @Test + public void testInvalidOrMissingHostNameFailsBeforeAnyCommand() { + FakeReconciler fake = new FakeReconciler(); + UserVm invalidName = vm("worker; reboot"); + UserVm noName = vm(null); + UserVm blankName = vm(" "); + + assertPhaseFailure("NODE_NAME", () -> fake.readNode(invalidName, access)); + assertPhaseFailure("NODE_NAME", () -> fake.readNode(noName, access)); + assertPhaseFailure("NODE_NAME", () -> fake.startResize(cluster, blankName, access, "uid-1")); + + Assert.assertTrue(fake.controlCommands.isEmpty()); + } + + @Test + public void testReadGuestParsesCpusMemTotalAndOnlineMemoryBlocks() { + FakeReconciler fake = new FakeReconciler(); + fake.guestOutput = "4\n4026532\n8000000\n32\n"; + + GuestResources guest = fake.readGuest(vm, access); + + Assert.assertEquals(4L, guest.getOnlineCpus()); + Assert.assertEquals(4026532L * 1024L, guest.getMemoryBytes()); + Assert.assertEquals(128L * MIB, guest.getMemoryBlockBytes()); + Assert.assertEquals(4L * GIB, guest.getOnlineMemoryBytes()); + Assert.assertEquals(List.of(KubernetesClusterNodeCapacityReconciler.GUEST_RESOURCES_COMMAND), fake.nodeCommands); + } + + @Test + public void testMalformedGuestOutputFails() { + FakeReconciler fake = new FakeReconciler(); + for (String output : List.of("4\n4194304", "0 4194304 8000000 32", "four 4194304 8000000 32", "4 4194304 8000000 32 1", + "4 4194304 0 32", "4 4194304 0x8000000 32", "4 4194304 8000000 0")) { + fake.guestOutput = output; + assertPhaseFailure("READ_GUEST", () -> fake.readGuest(vm, access)); + } + } + + @Test + public void testStartCordonsSchedulableNodeAndRecordsOwnershipInOnePatch() { + FakeReconciler fake = new FakeReconciler(); + + NodeCapacitySnapshot started = fake.startResize(cluster, vm, access, "uid-1"); + + Assert.assertEquals(List.of("sudo /opt/bin/kubectl patch node worker-1 --type merge --request-timeout=20s -p '{\"metadata\":" + + "{\"resourceVersion\":\"7\",\"annotations\":{\"" + PENDING + "\":\"cluster-uuid\",\"" + CORDON + "\":\"cluster-uuid\"}}," + + "\"spec\":{\"unschedulable\":true}}'"), fake.patches); + Assert.assertTrue(started.isUnschedulable()); + Assert.assertTrue(started.isPendingFor(CLUSTER_UUID)); + Assert.assertTrue(started.isCordonOwnedBy(CLUSTER_UUID)); + } + + @Test + public void testStartKeepsOperatorCordonAndAddsOnlyThePendingMarker() { + FakeReconciler fake = new FakeReconciler(); + fake.node.unschedulable = true; + + NodeCapacitySnapshot started = fake.startResize(cluster, vm, access, "uid-1"); + + Assert.assertEquals(1, fake.patches.size()); + Assert.assertFalse(fake.patches.get(0).contains("spec")); + Assert.assertFalse(fake.patches.get(0).contains(CORDON + "\"")); + Assert.assertTrue(started.isUnschedulable()); + Assert.assertTrue(started.isPendingFor(CLUSTER_UUID)); + Assert.assertFalse(started.isCordonOwnedBy(CLUSTER_UUID)); + } + + @Test + public void testStartDoesNothingWhenThePendingWorkIsAlreadyMarked() { + FakeReconciler fake = new FakeReconciler(); + fake.node.unschedulable = true; + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot started = fake.startResize(cluster, vm, access, "uid-1"); + + Assert.assertTrue(fake.patches.isEmpty()); + Assert.assertTrue(started.isCordonOwnedBy(CLUSTER_UUID)); + } + + @Test + public void testStartNeverChangesMarkersOfAnotherCluster() { + FakeReconciler fake = new FakeReconciler(); + fake.node.annotations.put(PENDING, "another-cluster"); + + assertPhaseFailure("START", () -> fake.startResize(cluster, vm, access, "uid-1")); + + Assert.assertTrue(fake.patches.isEmpty()); + } + + @Test + public void testRejectedPatchFailsAfterBoundedAttempts() { + FakeReconciler fake = new FakeReconciler(); + fake.failingPatches = 10; + + assertPhaseFailure("START", () -> fake.startResize(cluster, vm, access, "uid-1")); + + Assert.assertEquals(KubernetesClusterNodeCapacityReconciler.MAX_NODE_UPDATE_ATTEMPTS, fake.patches.size()); + Assert.assertFalse(fake.node.annotations.containsKey(PENDING)); + Assert.assertFalse(fake.node.unschedulable); + } + + @Test + public void testLostPatchReplyIsSettledByReadingTheNode() { + FakeReconciler fake = new FakeReconciler(); + fake.loseNextReply = true; + + NodeCapacitySnapshot started = fake.startResize(cluster, vm, access, "uid-1"); + + Assert.assertEquals(1, fake.patches.size()); + Assert.assertTrue(started.isPendingFor(CLUSTER_UUID)); + Assert.assertTrue(started.isUnschedulable()); + } + + @Test + public void testConflictIsRetriedWithTheNewResourceVersion() { + FakeReconciler fake = new FakeReconciler(); + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + fake.node.unschedulable = true; + // kubelet writes the node status right after its restart + fake.beforeNextPatch = () -> fake.node.resourceVersion++; + + NodeCapacitySnapshot finished = fake.finishResize(cluster, vm, access, "uid-1"); + + Assert.assertEquals(2, fake.patches.size()); + Assert.assertTrue(fake.patches.get(0).contains("\"resourceVersion\":\"7\"")); + Assert.assertTrue(fake.patches.get(1).contains("\"resourceVersion\":\"8\"")); + Assert.assertFalse(finished.isUnschedulable()); + Assert.assertFalse(finished.isPendingFor(CLUSTER_UUID)); + } + + @Test + public void testReplacedNodeStopsTheUpdate() { + FakeReconciler fake = new FakeReconciler(); + + assertPhaseFailure("FINISH", () -> fake.finishResize(cluster, vm, access, "uid-0")); + + Assert.assertTrue(fake.patches.isEmpty()); + } + + @Test + public void testFinishRemovesMarkersAndUncordonsOnlyACksCordon() { + FakeReconciler fake = new FakeReconciler(); + fake.node.unschedulable = true; + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + fake.node.annotations.put("other", "kept"); + + NodeCapacitySnapshot finished = fake.finishResize(cluster, vm, access, "uid-1"); + + Assert.assertEquals(1, fake.patches.size()); + Assert.assertTrue(fake.patches.get(0).contains("\"spec\":{\"unschedulable\":false}")); + Assert.assertFalse(finished.isUnschedulable()); + Assert.assertEquals(Map.of("other", "kept"), fake.node.annotations); + } + + @Test + public void testFinishKeepsAnOperatorCordon() { + FakeReconciler fake = new FakeReconciler(); + fake.node.unschedulable = true; + fake.node.annotations.put(PENDING, CLUSTER_UUID); + + NodeCapacitySnapshot finished = fake.finishResize(cluster, vm, access, "uid-1"); + + Assert.assertFalse(fake.patches.get(0).contains("spec")); + Assert.assertTrue(finished.isUnschedulable()); + Assert.assertFalse(finished.isPendingFor(CLUSTER_UUID)); + } + + @Test + public void testFinishDoesNothingWithoutMarkersOfThisCluster() { + FakeReconciler fake = new FakeReconciler(); + fake.node.annotations.put(PENDING, "another-cluster"); + + fake.finishResize(cluster, vm, access, "uid-1"); + + Assert.assertTrue(fake.patches.isEmpty()); + Assert.assertEquals("another-cluster", fake.node.annotations.get(PENDING)); + } + + @Test + public void testReleaseCordonKeepsThePendingMarker() { + FakeReconciler fake = new FakeReconciler(); + fake.node.unschedulable = true; + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot released = fake.releaseCordon(cluster, vm, access, "uid-1"); + + Assert.assertFalse(released.isUnschedulable()); + Assert.assertFalse(released.isCordonOwnedBy(CLUSTER_UUID)); + Assert.assertTrue(released.isPendingFor(CLUSTER_UUID)); + } + + @Test + public void testOwnershipMarkerOnASchedulableNodeIsNotACksCordon() { + FakeNode node = new FakeNode(); + node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot snapshot = KubernetesClusterNodeCapacityReconciler.parseNode(node.toJson()); + + Assert.assertTrue(snapshot.hasOwnershipMarker(CLUSTER_UUID)); + Assert.assertFalse(snapshot.isCordonOwnedBy(CLUSTER_UUID)); + } + + @Test + public void testFinishRemovesALeftoverOwnershipMarkerWithoutTouchingScheduling() { + FakeReconciler fake = new FakeReconciler(); + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot finished = fake.finishResize(cluster, vm, access, "uid-1"); + + Assert.assertFalse(fake.patches.get(0).contains("spec")); + Assert.assertFalse(finished.hasOwnershipMarker(CLUSTER_UUID)); + Assert.assertFalse(finished.isUnschedulable()); + } + + @Test + public void testReleaseRemovesALeftoverOwnershipMarkerWithoutTouchingScheduling() { + FakeReconciler fake = new FakeReconciler(); + fake.node.annotations.put(PENDING, CLUSTER_UUID); + fake.node.annotations.put(CORDON, CLUSTER_UUID); + + NodeCapacitySnapshot released = fake.releaseCordon(cluster, vm, access, "uid-1"); + + Assert.assertFalse(fake.patches.get(0).contains("spec")); + Assert.assertFalse(released.hasOwnershipMarker(CLUSTER_UUID)); + Assert.assertTrue(released.isPendingFor(CLUSTER_UUID)); + } + + @Test + public void testAlreadyResizedNeedsReadyNodeMatchingGuestAndAllocatableWithinCapacity() { + GuestResources guest = guest(4, 4L * 1024L * 1024L, 32); + + Assert.assertTrue(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 4000, 4 * GIB, 3900, 3 * GIB), guest)); + // reservations may use up all of the capacity + Assert.assertTrue(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 4000, 4 * GIB, 0, 0), guest)); + Assert.assertFalse(reconciler.isKubernetesNodeAlreadyResized(snapshot(false, 4000, 4 * GIB, 3900, 3 * GIB), guest)); + Assert.assertFalse(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 2000, 4 * GIB, 1900, 3 * GIB), guest)); + Assert.assertFalse(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 4000, 2 * GIB, 3900, GIB), guest)); + Assert.assertFalse(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 4000, 4 * GIB, 4100, 3 * GIB), guest)); + Assert.assertFalse(reconciler.isKubernetesNodeAlreadyResized(snapshot(true, 4000, 4 * GIB, 3900, 5 * GIB), guest)); + } + + @Test + public void testGuestMemoryIsCountedInOnlineMemoryBlocks() { + ServiceOffering target = offering(4, 4608); + + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(4, 4568788L, 36), target)); + // 512 MiB of the hot-added memory is not online yet + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(4, 4044500L, 32), target)); + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(2, 4568788L, 36), target)); + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(4, 4699860L, 37), target)); + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.isGuestAboveTarget(guest(4, 4699860L, 37), target)); + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.isGuestAboveTarget(guest(6, 4568788L, 36), target)); + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.isGuestAboveTarget(guest(4, 4568788L, 36), target)); + } + + @Test + public void testGuestMemoryCheckDoesNotDependOnTheMemoryTheKernelKeeps() { + // An 8 GiB VM reports about 230 MiB less MemTotal than its RAM; its memory blocks still add up to 8 GiB. + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(4, 8152860L, 64), offering(4, 8192))); + // RAM that is not a whole number of blocks ends in a partly filled block, counted as a whole one. + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.guestMatchesTarget(guest(1, 950000L, 8), offering(1, 1000))); + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.isGuestAboveTarget(guest(1, 950000L, 8), offering(1, 1000))); + } + + @Test + public void testGuestVerificationWaitsForHotpluggedResources() { + FakeReconciler fake = new FakeReconciler(); + fake.guestOutputs.add("2\n2029380\n8000000\n16\n"); + fake.guestOutput = "4\n4026532\n8000000\n32\n"; + ServiceOffering target = offering(4, 4096); + + GuestResources guest = fake.waitForGuestResources(vm, target, access, 60000L); + + Assert.assertEquals(4L, guest.getOnlineCpus()); + Assert.assertEquals(List.of(KubernetesClusterNodeCapacityReconciler.POLL_INTERVAL_MS), fake.sleeps); + } + + @Test + public void testGuestAboveTheTargetFailsAtOnce() { + FakeReconciler fake = new FakeReconciler(); + fake.guestOutput = "4\n8152860\n8000000\n64\n"; + ServiceOffering target = offering(4, 4096); + + assertPhaseFailure("GUEST_VERIFY", () -> fake.waitForGuestResources(vm, target, access, 60000L)); + + Assert.assertTrue(fake.sleeps.isEmpty()); + } + + @Test + public void testGuestVerificationStopsAtTheDeadline() { + FakeReconciler fake = new FakeReconciler(); + ServiceOffering target = offering(4, 4096); + + assertPhaseFailure("GUEST_VERIFY", () -> fake.waitForGuestResources(vm, target, access, 12000L)); + + Assert.assertEquals(List.of(5000L, 5000L, 2000L), fake.sleeps); + Assert.assertEquals(4, fake.nodeCommands.size()); + } + + @Test + public void testExpiredDeadlineAllowsOneObservationWithoutSleeping() { + FakeReconciler fake = new FakeReconciler(); + fake.now = 1000L; + GuestResources guest = guest(4, 4194304L, 32); + + assertPhaseFailure("CAPACITY_VERIFY", () -> fake.waitForKubernetesResources(vm, guest, access, 999L)); + + Assert.assertEquals(1, fake.controlCommands.size()); + Assert.assertTrue(fake.sleeps.isEmpty()); + } + + @Test + public void testWaitForKubernetesResourcesReturnsTheUpdatedNode() { + FakeReconciler fake = new FakeReconciler(); + fake.node.capacityCpu = "4"; + fake.node.capacityMemory = "4Gi"; + fake.node.allocatableCpu = "3900m"; + + NodeCapacitySnapshot node = fake.waitForKubernetesResources(vm, guest(4, 4194304L, 32), access, 60000L); + + Assert.assertEquals(4000L, node.getCapacity().getCpuMillis()); + Assert.assertTrue(fake.sleeps.isEmpty()); + } + + @Test + public void testRestartKubeletWaitsUntilActive() { + FakeReconciler fake = new FakeReconciler(); + fake.kubeletStates.add("activating"); + + fake.restartKubelet(vm, access, 60000L); + + Assert.assertEquals(List.of(KubernetesClusterNodeCapacityReconciler.RESTART_KUBELET_COMMAND, + KubernetesClusterNodeCapacityReconciler.KUBELET_STATE_COMMAND, + KubernetesClusterNodeCapacityReconciler.KUBELET_STATE_COMMAND), fake.nodeCommands); + Assert.assertEquals(1, fake.sleeps.size()); + } + + @Test + public void testKubeletIsNotRestartedAfterTheDeadline() { + FakeReconciler fake = new FakeReconciler(); + fake.now = 1000L; + + assertPhaseFailure("KUBELET_RESTART", () -> fake.restartKubelet(vm, access, 1000L)); + + Assert.assertTrue(fake.nodeCommands.isEmpty()); + } + + @Test + public void testInactiveKubeletIsReportedWithoutFailingTheCommand() { + FakeReconciler fake = new FakeReconciler(); + fake.kubeletStates.add("inactive"); + + Assert.assertFalse(fake.isKubeletActive(access)); + Assert.assertTrue(fake.isKubeletActive(access)); + Assert.assertTrue(KubernetesClusterNodeCapacityReconciler.KUBELET_STATE_COMMAND.endsWith("|| true")); + Assert.assertFalse(KubernetesClusterNodeCapacityReconciler.KUBELET_STATE_COMMAND.contains("sudo")); + } + + @Test + public void testOnlineGuestResourcesActivatesHotpluggedCpuAndMemory() { + FakeReconciler fake = new FakeReconciler(); + + fake.onlineGuestResources(access); + + Assert.assertEquals(List.of(KubernetesClusterNodeCapacityReconciler.ONLINE_HOTPLUGGED_RESOURCES), fake.nodeCommands); + } + + @Test + public void testNodeAccessKeepsControlAndNodeEndpointsApart() { + File key = new File("test-only-key"); + NodeAccess endpoints = new NodeAccess("control", 2222, "worker", 2223, "cloud", key); + + Assert.assertEquals("control", endpoints.getControlAddress()); + Assert.assertEquals(2222, endpoints.getControlPort()); + Assert.assertEquals("worker", endpoints.getNodeAddress()); + Assert.assertEquals(2223, endpoints.getNodePort()); + Assert.assertEquals("cloud", endpoints.getUser()); + Assert.assertSame(key, endpoints.getSshKeyFile()); + } + + @Test + public void testSshSuccessReturnsOutput() { + try (MockedStatic ssh = Mockito.mockStatic(SshHelper.class)) { + ssh.when(() -> SshHelper.sshExecute("node", 22, "root", null, null, "command", 10000, 10000, 30000)) + .thenReturn(new Pair<>(true, "output")); + Assert.assertEquals("output", reconciler.executeNode(access, "command")); + } + } + + @Test + public void testFailedSshCommandReportsItsOutput() { + try (MockedStatic ssh = Mockito.mockStatic(SshHelper.class)) { + ssh.when(() -> SshHelper.sshExecute("node", 22, "root", null, null, "command", 10000, 10000, 30000)) + .thenReturn(new Pair<>(false, "remote error")); + NodeCapacityReconciliationException error = Assert.assertThrows(NodeCapacityReconciliationException.class, + () -> reconciler.executeNode(access, "command")); + Assert.assertTrue(error.getMessage().contains("remote error")); + } + } + + @Test + public void testSshExceptionIsTranslatedAtTheTransportBoundary() { + Exception failure = new Exception("SSH authentication failed"); + try (MockedStatic ssh = Mockito.mockStatic(SshHelper.class)) { + ssh.when(() -> SshHelper.sshExecute("node", 22, "root", null, null, "command", 10000, 10000, 30000)) + .thenThrow(failure); + NodeCapacityReconciliationException error = Assert.assertThrows(NodeCapacityReconciliationException.class, + () -> reconciler.executeNode(access, "command")); + Assert.assertSame(failure, error.getCause()); + } + } + + @Test + public void testSshInterruptionKeepsTheInterruptFlag() { + InterruptedException failure = new InterruptedException("cancelled"); + try (MockedStatic ssh = Mockito.mockStatic(SshHelper.class)) { + ssh.when(() -> SshHelper.sshExecute("node", 22, "root", null, null, "command", 10000, 10000, 30000)) + .thenThrow(failure); + NodeCapacityReconciliationException error = Assert.assertThrows(NodeCapacityReconciliationException.class, + () -> reconciler.executeNode(access, "command")); + Assert.assertSame(failure, error.getCause()); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } + + @Test + public void testSshProgrammingErrorsAreNotTranslated() { + NullPointerException failure = new NullPointerException("programming error"); + try (MockedStatic ssh = Mockito.mockStatic(SshHelper.class)) { + ssh.when(() -> SshHelper.sshExecute("node", 22, "root", null, null, "command", 10000, 10000, 30000)) + .thenThrow(failure); + Assert.assertSame(failure, Assert.assertThrows(NullPointerException.class, + () -> reconciler.executeNode(access, "command"))); + } + } + + @Test + public void testInterruptedPollingKeepsTheInterruptFlag() { + KubernetesClusterNodeCapacityReconciler inactiveKubelet = new KubernetesClusterNodeCapacityReconciler() { + @Override + protected String executeNode(NodeAccess nodeAccess, String command) { + return "inactive"; + } + }; + long deadline = System.currentTimeMillis() + 10000L; + Thread.currentThread().interrupt(); + try { + NodeCapacityReconciliationException error = Assert.assertThrows(NodeCapacityReconciliationException.class, + () -> inactiveKubelet.restartKubelet(vm, access, deadline)); + Assert.assertTrue(error.getCause() instanceof InterruptedException); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } + + private void assertPhaseFailure(String phase, ThrowingRunnable action) { + NodeCapacityReconciliationException error = Assert.assertThrows(NodeCapacityReconciliationException.class, action); + Assert.assertTrue(error.getMessage(), error.getMessage().contains(phase + " failed for VM vm-uuid")); + } + + private static NodeCapacitySnapshot snapshot(boolean ready, long capacityCpuMillis, long capacityMemory, + long allocatableCpuMillis, long allocatableMemory) { + return new NodeCapacitySnapshot("uid-1", + "7", + false, + new HashMap<>(), + ready, + new NodeResources(capacityCpuMillis, capacityMemory), + new NodeResources(allocatableCpuMillis, allocatableMemory)); + } + + private static GuestResources guest(long cpus, long memTotalKiB, long onlineMemoryBlocks) { + return new GuestResources(cpus, memTotalKiB, onlineMemoryBlocks * 128L * MIB, 128L * MIB); + } + + private static ServiceOffering offering(int cpu, int memory) { + ServiceOffering offering = Mockito.mock(ServiceOffering.class); + Mockito.when(offering.getCpu()).thenReturn(cpu); + Mockito.when(offering.getRamSize()).thenReturn(memory); + return offering; + } + + private static KubernetesCluster cluster() { + KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(cluster.getUuid()).thenReturn(CLUSTER_UUID); + return cluster; + } + + private static UserVm vm(String hostname) { + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getHostName()).thenReturn(hostname); + Mockito.when(vm.getUuid()).thenReturn("vm-uuid"); + return vm; + } + + /** A Node as kubectl returns it, changed by the merge patches the reconciler sends. */ + private static class FakeNode { + private final Map annotations = new LinkedHashMap<>(); + private final String uid = "uid-1"; + private int resourceVersion = 7; + private boolean unschedulable; + private String capacityCpu = "2"; + private String capacityMemory = "2Gi"; + private String allocatableCpu = "1900m"; + private String allocatableMemory = "1900Mi"; + + String toJson() { + JsonObject metadata = new JsonObject(); + metadata.addProperty("name", "worker-1"); + metadata.addProperty("uid", uid); + metadata.addProperty("resourceVersion", String.valueOf(resourceVersion)); + JsonObject annotationsJson = new JsonObject(); + annotations.forEach(annotationsJson::addProperty); + metadata.add("annotations", annotationsJson); + JsonObject spec = new JsonObject(); + if (unschedulable) { + spec.addProperty("unschedulable", true); + } + JsonObject capacity = new JsonObject(); + capacity.addProperty("cpu", capacityCpu); + capacity.addProperty("memory", capacityMemory); + JsonObject allocatable = new JsonObject(); + allocatable.addProperty("cpu", allocatableCpu); + allocatable.addProperty("memory", allocatableMemory); + JsonObject readyCondition = new JsonObject(); + readyCondition.addProperty("type", "Ready"); + readyCondition.addProperty("status", "True"); + JsonArray conditions = new JsonArray(); + conditions.add(readyCondition); + JsonObject status = new JsonObject(); + status.add("capacity", capacity); + status.add("allocatable", allocatable); + status.add("conditions", conditions); + JsonObject node = new JsonObject(); + node.add("metadata", metadata); + node.add("spec", spec); + node.add("status", status); + return node.toString(); + } + } + + private static class FakeReconciler extends KubernetesClusterNodeCapacityReconciler { + private final FakeNode node = new FakeNode(); + private final List controlCommands = new ArrayList<>(); + private final List nodeCommands = new ArrayList<>(); + private final List patches = new ArrayList<>(); + private final List sleeps = new ArrayList<>(); + private final Deque guestOutputs = new ArrayDeque<>(); + private final Deque kubeletStates = new ArrayDeque<>(); + private String guestOutput = "2\n2029380\n8000000\n16\n"; + private String nodeOutput; + private int failingPatches; + private boolean loseNextReply; + private Runnable beforeNextPatch; + private long now; + + @Override + protected String executeControl(NodeAccess nodeAccess, String command) { + controlCommands.add(command); + if (command.contains(" patch node ")) { + return patch(command); + } + return nodeOutput != null ? nodeOutput : node.toJson(); + } + + @Override + protected String executeNode(NodeAccess nodeAccess, String command) { + nodeCommands.add(command); + if (KubernetesClusterNodeCapacityReconciler.GUEST_RESOURCES_COMMAND.equals(command)) { + return guestOutputs.isEmpty() ? guestOutput : guestOutputs.poll(); + } + if (KubernetesClusterNodeCapacityReconciler.KUBELET_STATE_COMMAND.equals(command)) { + return kubeletStates.isEmpty() ? "active\n" : kubeletStates.poll(); + } + return ""; + } + + @Override + protected long currentTimeMillis() { + return now; + } + + @Override + protected void sleep(long millis) { + sleeps.add(millis); + now += millis; + } + + private String patch(String command) { + patches.add(command); + if (beforeNextPatch != null) { + Runnable change = beforeNextPatch; + beforeNextPatch = null; + change.run(); + } + if (failingPatches > 0) { + failingPatches--; + throw new NodeCapacityReconciliationException("Error from server (Forbidden)"); + } + JsonObject patch = JsonParser.parseString(command.substring(command.indexOf("-p '") + 4, command.length() - 1)).getAsJsonObject(); + JsonObject metadata = patch.getAsJsonObject("metadata"); + if (!String.valueOf(node.resourceVersion).equals(metadata.get("resourceVersion").getAsString())) { + throw new NodeCapacityReconciliationException("Error from server (Conflict): the object has been modified"); + } + for (Map.Entry annotation : metadata.getAsJsonObject("annotations").entrySet()) { + if (annotation.getValue().isJsonNull()) { + node.annotations.remove(annotation.getKey()); + } else { + node.annotations.put(annotation.getKey(), annotation.getValue().getAsString()); + } + } + if (patch.has("spec")) { + node.unschedulable = patch.getAsJsonObject("spec").get("unschedulable").getAsBoolean(); + } + node.resourceVersion++; + if (loseNextReply) { + loseNextReply = false; + throw new NodeCapacityReconciliationException("Unable to run command: connection reset"); + } + return "node/worker-1 patched"; + } + } +} diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java index 329f9b0e42ad..e0fc2b8d5b84 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtilTest.java @@ -16,10 +16,12 @@ // under the License. package com.cloud.kubernetes.cluster.utils; +import com.cloud.uservm.UserVm; import com.cloud.utils.Pair; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; @RunWith(MockitoJUnitRunner.class) @@ -52,4 +54,20 @@ public void testClusterNodeVersionMatches() { executeThrowAndTestVersionMatch(); } + + @Test + public void testKubernetesNodeNameIsTheLowercaseHostName() { + UserVm vm = Mockito.mock(UserVm.class); + Mockito.when(vm.getHostName()).thenReturn("Worker-1"); + Assert.assertEquals("worker-1", KubernetesClusterUtil.getKubernetesNodeName(vm)); + Mockito.when(vm.getHostName()).thenReturn(null); + Assert.assertNull(KubernetesClusterUtil.getKubernetesNodeName(vm)); + } + + @Test + public void testNodePatchCommandAppliesOneMergePatch() { + Assert.assertEquals("sudo /opt/bin/kubectl patch node worker-1 --type merge --request-timeout=20s -p '{\"spec\":{\"unschedulable\":true}}'", + KubernetesClusterUtil.getKubernetesNodePatchCommand("worker-1", "{\"spec\":{\"unschedulable\":true}}")); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterUtil.getKubernetesNodePatchCommand("worker-1", "{\"a\":\"'\"}")); + } }