From a62d23b0bc4ae92e00899abd14a465562650365d Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 12:47:24 +0200 Subject: [PATCH 01/11] CKS: reconcile node capacity after live resize --- .../KubernetesClusterScaleWorker.java | 99 ++++++++- ...bernetesClusterNodeCapacityReconciler.java | 67 ++++++ ...etesClusterNodeCapacityReconcilerImpl.java | 190 ++++++++++++++++++ .../spring-kubernetes-service-context.xml | 1 + .../KubernetesClusterScaleWorkerTest.java | 33 +++ ...ClusterNodeCapacityReconcilerImplTest.java | 71 +++++++ 6 files changed, 455 insertions(+), 6 deletions(-) create mode 100644 plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java create mode 100644 plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java create mode 100644 plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java 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..2df102531c66 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 @@ -28,6 +28,8 @@ import java.util.Set; import java.util.stream.Collectors; +import javax.inject.Inject; + import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; import com.cloud.service.ServiceOfferingVO; import com.cloud.storage.VMTemplateVO; @@ -44,16 +46,21 @@ import com.cloud.exception.ManagementServerException; import com.cloud.exception.NetworkRuleConflictException; 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.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.KubernetesClusterNodeCapacityReconciler.NodeAccess; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconcilerImpl; 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; @@ -80,6 +87,9 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif private Boolean isAutoscalingEnabled; private long scaleTimeoutTime; + @Inject + protected KubernetesClusterNodeCapacityReconciler kubernetesClusterNodeCapacityReconciler; + protected KubernetesClusterScaleWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { super(kubernetesCluster, clusterManager); } @@ -378,15 +388,60 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, for (long i = 0; i < tobeScaledVMCount; i++) { KubernetesClusterVmMapVO vmMapVO = vmList.get((int) i); UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); + if (userVM == null) { + logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster : %s failed, unable to find cluster VM %s", + kubernetesCluster.getName(), vmMapVO.getVmId()), kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + ServiceOffering oldOffering = serviceOfferingDao.findById(userVM.getServiceOfferingId()); + boolean capacityChanged = KubernetesClusterNodeCapacityReconcilerImpl.capacityChanged(oldOffering, serviceOffering); + if (capacityChanged && KubernetesCluster.State.Running == originalState && ETCD == nodeType + && KubernetesCluster.ClusterType.CloudManaged == kubernetesCluster.getClusterType()) { + logTransitStateAndThrow(Level.ERROR, String.format("Scaling Kubernetes cluster : %s cannot live-resize dedicated etcd VM %s; " + + "etcd health reconciliation is not implemented", kubernetesCluster.getName(), userVM.getDisplayName()), + kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + boolean reconcileKubernetes = shouldReconcileNodeCapacity(vmMapVO, userVM, oldOffering, serviceOffering, nodeType); + NodeAccess nodeAccess = null; + NodeCapacitySnapshot before = null; boolean result = false; + boolean cksCordonCompleted = false; + boolean resizeSucceeded = serviceOffering.getId() == userVM.getServiceOfferingId(); try { - result = userVmManager.upgradeVirtualMachine(userVM.getId(), serviceOffering.getId(), new HashMap()); - } catch (RuntimeException | ResourceUnavailableException | ManagementServerException | VirtualMachineMigrationException e) { + if (reconcileKubernetes) { + nodeAccess = resolveNodeAccess(userVM); + before = kubernetesClusterNodeCapacityReconciler.captureBefore(kubernetesCluster, userVM, nodeAccess); + kubernetesClusterNodeCapacityReconciler.cordonIfNeeded(kubernetesCluster, userVM, before, nodeAccess, scaleTimeoutTime); + cksCordonCompleted = !before.isUnschedulable() || before.isCloudStackResizeCordon(); + } + if (serviceOffering.getId() != userVM.getServiceOfferingId()) { + result = userVmManager.upgradeVirtualMachine(userVM.getId(), serviceOffering.getId(), new HashMap()); + 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); + } + resizeSucceeded = true; + userVM = userVmDao.findById(userVM.getId()); + if (userVM == null || serviceOffering.getId() != userVM.getServiceOfferingId()) { + logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster : %s failed, VM %s did not reach target offering", kubernetesCluster.getName(), vmMapVO.getVmId()),kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); + } + } + if (reconcileKubernetes) { + kubernetesClusterNodeCapacityReconciler.verifyGuestResources(userVM, serviceOffering, before, nodeAccess, scaleTimeoutTime); + if (!kubernetesClusterNodeCapacityReconciler.isKubernetesResourcesCurrent(before, serviceOffering)) { + kubernetesClusterNodeCapacityReconciler.restartKubelet(userVM, nodeAccess, scaleTimeoutTime); + kubernetesClusterNodeCapacityReconciler.waitForKubernetesResources(kubernetesCluster, userVM, serviceOffering, before, nodeAccess, scaleTimeoutTime); + } + kubernetesClusterNodeCapacityReconciler.restoreSchedulability(kubernetesCluster, userVM, before, nodeAccess, scaleTimeoutTime); + } + } catch (Exception e) { + if (reconcileKubernetes && cksCordonCompleted && !resizeSucceeded) { + try { + kubernetesClusterNodeCapacityReconciler.restoreSchedulability(kubernetesCluster, userVM, before, nodeAccess, scaleTimeoutTime); + } catch (Exception cleanupException) { + logger.warn("Unable to restore schedulability for CKS node {} after a pre-resize failure", userVM.getUuid(), cleanupException); + } + } 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); - } 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); } @@ -394,6 +449,38 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOffering, updateNodeOffering, updateClusterOffering); } + private 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()); + } + PortForwardingRuleVO sshRule = portForwardingRulesDao.listByVm(userVM.getId()).stream() + .filter(rule -> rule.getDestinationPortStart() == DEFAULT_SSH_PORT) + .filter(rule -> !FirewallRule.State.Revoke.equals(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())); + } + return new NodeAccess(controlAccess.first(), controlAccess.second(), controlAccess.first(), sshRule.getSourcePortStart(), + getControlNodeLoginUser(), getManagementServerSshPublicKeyFile()); + } + + protected boolean shouldReconcileNodeCapacity(KubernetesClusterVmMapVO vmMapVO, UserVm userVM, + ServiceOffering oldOffering, ServiceOffering targetOffering, + KubernetesClusterNodeType nodeType) { + return !vmMapVO.isExternalNode() + && KubernetesCluster.ClusterType.CloudManaged == kubernetesCluster.getClusterType() + && Hypervisor.HypervisorType.KVM == userVM.getHypervisorType() + && kubernetesClusterNodeCapacityReconciler.requiresKubeletRefresh(oldOffering, targetOffering, nodeType, originalState); + } + private void removeNodesFromCluster(List vmMaps) throws CloudRuntimeException { for (KubernetesClusterVmMapVO vmMapVO : vmMaps) { UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); 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..f52458774aca --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.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; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; +import com.cloud.offering.ServiceOffering; +import com.cloud.uservm.UserVm; + +/** Reconciles the guest and Kubernetes resource views after a live VM resize. */ +public interface KubernetesClusterNodeCapacityReconciler { + NodeCapacitySnapshot captureBefore(KubernetesCluster cluster, UserVm vm, NodeAccess access) throws Exception; + boolean requiresKubeletRefresh(ServiceOffering oldOffering, ServiceOffering newOffering, + KubernetesClusterNodeType nodeType, KubernetesCluster.State originalState); + void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; + void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; + boolean isKubernetesResourcesCurrent(NodeCapacitySnapshot snapshot, ServiceOffering target); + void restartKubelet(UserVm vm, NodeAccess access, long deadline) throws Exception; + NodeCapacitySnapshot waitForKubernetesResources(KubernetesCluster cluster, UserVm vm, ServiceOffering target, + NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; + void restoreSchedulability(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; + + 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; } + } + + final class NodeCapacitySnapshot { + private final boolean unschedulable; private final boolean cloudStackResizeCordon; private final long guestOnlineCpuCount; + private final long guestMemoryKiB; private final long capacityCpuMillis; private final long capacityMemoryBytes; + private final long allocatableCpuMillis; private final long allocatableMemoryBytes; private final boolean ready; + public NodeCapacitySnapshot(boolean unschedulable, boolean cloudStackResizeCordon, long guestOnlineCpuCount, long guestMemoryKiB, + long capacityCpuMillis, long capacityMemoryBytes, long allocatableCpuMillis, long allocatableMemoryBytes, boolean ready) { + this.unschedulable = unschedulable; this.cloudStackResizeCordon = cloudStackResizeCordon; this.guestOnlineCpuCount = guestOnlineCpuCount; + this.guestMemoryKiB = guestMemoryKiB; this.capacityCpuMillis = capacityCpuMillis; this.capacityMemoryBytes = capacityMemoryBytes; + this.allocatableCpuMillis = allocatableCpuMillis; this.allocatableMemoryBytes = allocatableMemoryBytes; this.ready = ready; + } + public boolean isUnschedulable() { return unschedulable; } public boolean isCloudStackResizeCordon() { return cloudStackResizeCordon; } + public long getGuestOnlineCpuCount() { return guestOnlineCpuCount; } public long getGuestMemoryKiB() { return guestMemoryKiB; } + public long getCapacityCpuMillis() { return capacityCpuMillis; } public long getCapacityMemoryBytes() { return capacityMemoryBytes; } + public long getAllocatableCpuMillis() { return allocatableCpuMillis; } public long getAllocatableMemoryBytes() { return allocatableMemoryBytes; } + public boolean isReady() { return ready; } + } +} diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java new file mode 100644 index 000000000000..a6785ec3fdfd --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java @@ -0,0 +1,190 @@ +// 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.Objects; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import com.cloud.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; +import com.cloud.offering.ServiceOffering; +import com.cloud.uservm.UserVm; +import com.cloud.utils.Pair; +import com.cloud.utils.exception.CloudRuntimeException; +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; +import org.apache.commons.lang3.StringUtils; + +/** SSH-backed implementation for a rolling CKS live-resize reconciliation. */ +public class KubernetesClusterNodeCapacityReconcilerImpl implements KubernetesClusterNodeCapacityReconciler { + static final long MINIMUM_MEMORY_OVERHEAD_BYTES = 128L * 1024L * 1024L; + static final long KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES = 16L * 1024L * 1024L; + private static final int COMMAND_TIMEOUT_MS = 30000; + private static final int POLL_INTERVAL_MS = 5000; + private static final Pattern NODE_NAME = Pattern.compile("[A-Za-z0-9][A-Za-z0-9.-]{0,252}"); + private static final Pattern MEMORY_QUANTITY = Pattern.compile("([0-9]+)(Ki|Mi|Gi|Ti|K|M|G|T)?"); + private static final String RESIZE_ANNOTATION = "cloudstack.apache.org/cks-live-resize"; + + @Override + public boolean requiresKubeletRefresh(ServiceOffering oldOffering, ServiceOffering newOffering, + KubernetesClusterNodeType nodeType, KubernetesCluster.State originalState) { + return KubernetesCluster.State.Running == originalState + && (KubernetesClusterNodeType.WORKER == nodeType || KubernetesClusterNodeType.CONTROL == nodeType) + && capacityChanged(oldOffering, newOffering); + } + + @Override + public NodeCapacitySnapshot captureBefore(KubernetesCluster cluster, UserVm vm, NodeAccess access) throws Exception { + return nodeSnapshot(cluster, vm, access, guestResources(access)); + } + + @Override + public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception { + if (before.isUnschedulable() && !before.isCloudStackResizeCordon()) return; + String node = nodeName(vm); + executeControl(access, "sudo /opt/bin/kubectl annotate node " + node + " " + RESIZE_ANNOTATION + "=" + cluster.getUuid() + " --overwrite"); + executeControl(access, "sudo /opt/bin/kubectl cordon " + node); + while (System.currentTimeMillis() < deadline) { + if (nodeSnapshot(cluster, vm, access, null).isUnschedulable()) return; + sleep(); + } + throw failure("CORDON", vm, "Kubernetes node did not become unschedulable"); + } + + @Override + public void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception { + while (System.currentTimeMillis() < deadline) { + if (guestMatchesTarget(guestResources(access), target)) return; + sleep(); + } + throw failure("GUEST_VERIFY", vm, "guest CPU or memory did not reach the target offering"); + } + + @Override + public boolean isKubernetesResourcesCurrent(NodeCapacitySnapshot snapshot, ServiceOffering target) { + GuestResources guest = new GuestResources(snapshot.getGuestOnlineCpuCount(), snapshot.getGuestMemoryKiB()); + return guestMatchesTarget(guest, target) && kubernetesMatchesGuest(snapshot, snapshot); + } + + @Override + public void restartKubelet(UserVm vm, NodeAccess access, long deadline) throws Exception { + executeNode(access, "sudo systemctl restart kubelet"); + while (System.currentTimeMillis() < deadline) { + if ("active".equals(executeNode(access, "sudo systemctl is-active kubelet").trim())) return; + sleep(); + } + throw failure("KUBELET_RESTART", vm, "kubelet did not become active"); + } + + @Override + public NodeCapacitySnapshot waitForKubernetesResources(KubernetesCluster cluster, UserVm vm, ServiceOffering target, + NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception { + while (System.currentTimeMillis() < deadline) { + GuestResources guest = guestResources(access); + NodeCapacitySnapshot observed = nodeSnapshot(cluster, vm, access, guest); + if (guestMatchesTarget(guest, target) && kubernetesMatchesGuest(observed, before)) return observed; + sleep(); + } + throw failure("CAPACITY_VERIFY", vm, "Kubernetes capacity did not match the resized guest"); + } + + @Override + public void restoreSchedulability(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception { + if (before.isUnschedulable() && !before.isCloudStackResizeCordon()) return; + String node = nodeName(vm); + executeControl(access, "sudo /opt/bin/kubectl uncordon " + node); + executeControl(access, "sudo /opt/bin/kubectl annotate node " + node + " " + RESIZE_ANNOTATION + "-"); + while (System.currentTimeMillis() < deadline) { + if (!nodeSnapshot(cluster, vm, access, null).isUnschedulable()) return; + sleep(); + } + throw failure("RESTORE_SCHEDULABILITY", vm, "Kubernetes node remained unschedulable"); + } + + public static boolean capacityChanged(ServiceOffering oldOffering, ServiceOffering targetOffering) { + return oldOffering != null && targetOffering != null + && (!Objects.equals(oldOffering.getCpu(), targetOffering.getCpu()) + || !Objects.equals(oldOffering.getRamSize(), targetOffering.getRamSize())); + } + + static long parseCpuMillis(String value) { return value.endsWith("m") ? Long.parseLong(value.substring(0, value.length() - 1)) : Long.parseLong(value) * 1000L; } + static long parseMemoryBytes(String value) { + Matcher matcher = MEMORY_QUANTITY.matcher(value); + if (!matcher.matches()) throw new IllegalArgumentException("Unsupported Kubernetes memory quantity"); + long number = Long.parseLong(matcher.group(1)); String unit = matcher.group(2); + if (unit == null) return number; + switch (unit) { + case "Ki": return number * 1024L; case "Mi": return number * 1024L * 1024L; case "Gi": return number * 1024L * 1024L * 1024L; + case "Ti": return number * 1024L * 1024L * 1024L * 1024L; case "K": return number * 1000L; case "M": return number * 1000L * 1000L; + case "G": return number * 1000L * 1000L * 1000L; case "T": return number * 1000L * 1000L * 1000L * 1000L; + default: throw new IllegalArgumentException("Unsupported Kubernetes memory quantity"); + } + } + + private NodeCapacitySnapshot nodeSnapshot(KubernetesCluster cluster, UserVm vm, NodeAccess access, GuestResources guest) throws Exception { + JsonObject node = new JsonParser().parse(executeControl(access, "sudo /opt/bin/kubectl get node " + nodeName(vm) + " -o json")).getAsJsonObject(); + JsonObject annotations = node.getAsJsonObject("metadata").has("annotations") ? node.getAsJsonObject("metadata").getAsJsonObject("annotations") : null; + boolean cksCordon = annotations != null && annotations.has(RESIZE_ANNOTATION) && cluster.getUuid().equals(annotations.get(RESIZE_ANNOTATION).getAsString()); + JsonObject spec = node.has("spec") ? node.getAsJsonObject("spec") : new JsonObject(); JsonObject status = node.getAsJsonObject("status"); + JsonObject capacity = status.getAsJsonObject("capacity"); JsonObject allocatable = status.getAsJsonObject("allocatable"); + return new NodeCapacitySnapshot(spec.has("unschedulable") && spec.get("unschedulable").getAsBoolean(), cksCordon, + guest == null ? 0 : guest.cpu, guest == null ? 0 : guest.memoryKiB, parseCpuMillis(capacity.get("cpu").getAsString()), + parseMemoryBytes(capacity.get("memory").getAsString()), parseCpuMillis(allocatable.get("cpu").getAsString()), + parseMemoryBytes(allocatable.get("memory").getAsString()), isReady(status)); + } + + private boolean isReady(JsonObject status) { + JsonArray conditions = status.getAsJsonArray("conditions"); + for (JsonElement condition : conditions) { JsonObject item = condition.getAsJsonObject(); if ("Ready".equals(item.get("type").getAsString()) && "True".equals(item.get("status").getAsString())) return true; } + return false; + } + + private GuestResources guestResources(NodeAccess access) throws Exception { + String[] values = executeNode(access, "getconf _NPROCESSORS_ONLN; awk '/^MemTotal:/ {print $2}' /proc/meminfo").trim().split("\\s+"); + if (values.length != 2) throw new CloudRuntimeException("Unable to read guest CPU and memory"); + return new GuestResources(Long.parseLong(values[0]), Long.parseLong(values[1])); + } + + private boolean guestMatchesTarget(GuestResources observed, ServiceOffering target) { + long targetMemory = target.getRamSize() * 1024L * 1024L; long observedMemory = observed.memoryKiB * 1024L; + long overhead = Math.max(targetMemory / 50L, MINIMUM_MEMORY_OVERHEAD_BYTES); + return observed.cpu == target.getCpu() && observedMemory <= targetMemory && targetMemory - observedMemory <= overhead; + } + + private boolean kubernetesMatchesGuest(NodeCapacitySnapshot observed, NodeCapacitySnapshot before) { + return observed.isReady() && observed.getCapacityCpuMillis() == observed.getGuestOnlineCpuCount() * 1000L + && Math.abs(observed.getCapacityMemoryBytes() - observed.getGuestMemoryKiB() * 1024L) <= KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES + && (observed.getGuestOnlineCpuCount() <= before.getGuestOnlineCpuCount() || observed.getAllocatableCpuMillis() > before.getAllocatableCpuMillis()) + && (observed.getGuestMemoryKiB() <= before.getGuestMemoryKiB() || observed.getAllocatableMemoryBytes() > before.getAllocatableMemoryBytes()); + } + + private String executeControl(NodeAccess access, String command) throws Exception { return execute(access.getControlAddress(), access.getControlPort(), access, command); } + private String executeNode(NodeAccess access, String command) throws Exception { return execute(access.getNodeAddress(), access.getNodePort(), access, command); } + private String execute(String address, int port, NodeAccess access, String command) throws Exception { + Pair result = SshHelper.sshExecute(address, port, access.getUser(), access.getSshKeyFile(), null, command, 10000, 10000, COMMAND_TIMEOUT_MS); + if (Boolean.TRUE.equals(result.first())) return StringUtils.defaultString(result.second()); + throw new CloudRuntimeException("CKS live-resize command failed"); + } + private String nodeName(UserVm vm) { String name = StringUtils.lowerCase(vm.getHostName()); if (!NODE_NAME.matcher(name).matches()) throw new CloudRuntimeException("Invalid Kubernetes node name for live resize"); return name; } + private CloudRuntimeException failure(String phase, UserVm vm, String message) { return new CloudRuntimeException("CKS live resize " + phase + " failed for VM " + vm.getUuid() + ": " + message); } + private void sleep() throws InterruptedException { Thread.sleep(POLL_INTERVAL_MS); } + private static class GuestResources { private final long cpu; private final long memoryKiB; GuestResources(long cpu, long memoryKiB) { this.cpu = cpu; this.memoryKiB = memoryKiB; } } +} diff --git a/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml b/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml index 053366786292..9fe19200d6e7 100644 --- a/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml +++ b/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml @@ -34,6 +34,7 @@ + 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..ffd9a28288cc 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 @@ -20,6 +20,8 @@ import com.cloud.kubernetes.cluster.KubernetesClusterVmMapVO; import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler; +import com.cloud.hypervisor.Hypervisor; import com.cloud.offering.ServiceOffering; import com.cloud.service.ServiceOfferingVO; import com.cloud.service.dao.ServiceOfferingDao; @@ -40,6 +42,7 @@ 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.WORKER; @RunWith(MockitoJUnitRunner.class) public class KubernetesClusterScaleWorkerTest { @@ -54,6 +57,8 @@ public class KubernetesClusterScaleWorkerTest { private KubernetesClusterVmMapDao kubernetesClusterVmMapDao; @Mock private UserVmDao userVmDao; + @Mock + private KubernetesClusterNodeCapacityReconciler kubernetesClusterNodeCapacityReconciler; private KubernetesClusterScaleWorker worker; @@ -187,4 +192,32 @@ public void testGetWorkerNodesToRemoveForDownsize_noRemoval() { Assert.assertTrue(toRemove.isEmpty()); } + + @Test + public void testShouldReconcileNodeCapacityOnlyForManagedKvmNodes() { + KubernetesCluster runningManagedCluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(runningManagedCluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(runningManagedCluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.CloudManaged); + KubernetesClusterScaleWorker scaleWorker = new KubernetesClusterScaleWorker(runningManagedCluster, + new java.util.HashMap<>(), 1L, null, false, null, null, clusterManager); + scaleWorker.kubernetesClusterNodeCapacityReconciler = kubernetesClusterNodeCapacityReconciler; + + KubernetesClusterVmMapVO managedNode = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.when(managedNode.isExternalNode()).thenReturn(false); + UserVmVO kvmNode = Mockito.mock(UserVmVO.class); + Mockito.when(kvmNode.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM); + ServiceOffering oldOffering = Mockito.mock(ServiceOffering.class); + ServiceOffering targetOffering = Mockito.mock(ServiceOffering.class); + Mockito.when(kubernetesClusterNodeCapacityReconciler.requiresKubeletRefresh(oldOffering, targetOffering, WORKER, + KubernetesCluster.State.Running)).thenReturn(true); + + Assert.assertTrue(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); + + Mockito.when(managedNode.isExternalNode()).thenReturn(true); + Assert.assertFalse(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); + + Mockito.when(managedNode.isExternalNode()).thenReturn(false); + Mockito.when(kvmNode.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.XenServer); + Assert.assertFalse(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java new file mode 100644 index 000000000000..729803c7187e --- /dev/null +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -0,0 +1,71 @@ +// 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.kubernetes.cluster.KubernetesCluster; +import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; +import com.cloud.offering.ServiceOffering; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +public class KubernetesClusterNodeCapacityReconcilerImplTest { + private final KubernetesClusterNodeCapacityReconciler reconciler = new KubernetesClusterNodeCapacityReconcilerImpl(); + + @Test + public void testParseKubernetesQuantities() { + Assert.assertEquals(3900L, KubernetesClusterNodeCapacityReconcilerImpl.parseCpuMillis("3900m")); + Assert.assertEquals(4000L, KubernetesClusterNodeCapacityReconcilerImpl.parseCpuMillis("4")); + Assert.assertEquals(4037048L * 1024L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("4037048Ki")); + Assert.assertEquals(4L * 1024L * 1024L * 1024L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("4Gi")); + } + + @Test + public void testOnlyRunningKubernetesNodesWithChangedCpuOrRamRequireRefresh() { + ServiceOffering oldOffering = offering(2, 2048); + ServiceOffering cpuOffering = offering(4, 2048); + ServiceOffering ramOffering = offering(2, 4096); + ServiceOffering capOnlyOffering = offering(2, 2048); + + Assert.assertTrue(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); + Assert.assertTrue(reconciler.requiresKubeletRefresh(oldOffering, ramOffering, KubernetesClusterNodeType.CONTROL, KubernetesCluster.State.Running)); + Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, capOnlyOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); + Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.ETCD, KubernetesCluster.State.Running)); + Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Stopped)); + } + + @Test + public void testCurrentKubernetesResourcesDoNotRequireASecondKubeletRestart() { + long memoryKiB = 4L * 1024L * 1024L; + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot current = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, 4, memoryKiB, + 4000, memoryKiB * 1024L, 3900, memoryKiB * 1024L - 128L * 1024L * 1024L, true); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot stale = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, 4, memoryKiB, + 2000, memoryKiB * 1024L, 1900, memoryKiB * 1024L - 128L * 1024L * 1024L, true); + + Assert.assertTrue(reconciler.isKubernetesResourcesCurrent(current, offering(4, 4096))); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(stale, offering(4, 4096))); + } + + private 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; + } +} From 95a51822705c1eec650a2a58033f60e2f0038ebe Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 13:32:09 +0200 Subject: [PATCH 02/11] CKS: cover node capacity reconciliation ownership --- ...etesClusterNodeCapacityReconcilerImpl.java | 4 +- ...ClusterNodeCapacityReconcilerImplTest.java | 82 +++++++++++++++++++ 2 files changed, 84 insertions(+), 2 deletions(-) diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java index a6785ec3fdfd..68576ce12c9e 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java @@ -176,8 +176,8 @@ private boolean kubernetesMatchesGuest(NodeCapacitySnapshot observed, NodeCapaci && (observed.getGuestMemoryKiB() <= before.getGuestMemoryKiB() || observed.getAllocatableMemoryBytes() > before.getAllocatableMemoryBytes()); } - private String executeControl(NodeAccess access, String command) throws Exception { return execute(access.getControlAddress(), access.getControlPort(), access, command); } - private String executeNode(NodeAccess access, String command) throws Exception { return execute(access.getNodeAddress(), access.getNodePort(), access, command); } + protected String executeControl(NodeAccess access, String command) throws Exception { return execute(access.getControlAddress(), access.getControlPort(), access, command); } + protected String executeNode(NodeAccess access, String command) throws Exception { return execute(access.getNodeAddress(), access.getNodePort(), access, command); } private String execute(String address, int port, NodeAccess access, String command) throws Exception { Pair result = SshHelper.sshExecute(address, port, access.getUser(), access.getSshKeyFile(), null, command, 10000, 10000, COMMAND_TIMEOUT_MS); if (Boolean.TRUE.equals(result.first())) return StringUtils.defaultString(result.second()); diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java index 729803c7187e..3707c8014bff 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -16,9 +16,13 @@ // under the License. package com.cloud.kubernetes.cluster.utils; +import java.util.ArrayList; +import java.util.List; + import com.cloud.kubernetes.cluster.KubernetesCluster; import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; import com.cloud.offering.ServiceOffering; +import com.cloud.uservm.UserVm; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -62,10 +66,88 @@ public void testCurrentKubernetesResourcesDoNotRequireASecondKubeletRestart() { Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(stale, offering(4, 4096))); } + @Test + public void testCaptureBeforeReadsStructuredNodeState() throws Exception { + FakeReconciler fakeReconciler = new FakeReconciler(nodeJson(false)); + KubernetesCluster cluster = cluster("cluster-uuid"); + UserVm vm = vm("worker-1"); + + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot snapshot = fakeReconciler.captureBefore(cluster, vm, access()); + + Assert.assertFalse(snapshot.isUnschedulable()); + Assert.assertTrue(snapshot.isReady()); + Assert.assertEquals(2L, snapshot.getGuestOnlineCpuCount()); + Assert.assertEquals(2097152L, snapshot.getGuestMemoryKiB()); + Assert.assertEquals(3900L, snapshot.getCapacityCpuMillis()); + Assert.assertEquals(4L * 1024L * 1024L * 1024L, snapshot.getCapacityMemoryBytes()); + } + + @Test + public void testCordonAndRestoreOnlyNodesOwnedByCks() throws Exception { + KubernetesCluster cluster = cluster("cluster-uuid"); + UserVm vm = vm("worker-1"); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot previouslyCordon = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(true, false, 0, 0, 0, 0, 0, 0, true); + FakeReconciler operatorCordon = new FakeReconciler(nodeJson(true)); + + operatorCordon.cordonIfNeeded(cluster, vm, previouslyCordon, access(), System.currentTimeMillis() + 1000L); + operatorCordon.restoreSchedulability(cluster, vm, previouslyCordon, access(), System.currentTimeMillis() + 1000L); + Assert.assertTrue(operatorCordon.controlCommands.isEmpty()); + + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot cksCordon = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(true, true, 0, 0, 0, 0, 0, 0, true); + FakeReconciler cleanup = new FakeReconciler(nodeJson(false)); + cleanup.restoreSchedulability(cluster, vm, cksCordon, access(), System.currentTimeMillis() + 1000L); + + Assert.assertEquals("sudo /opt/bin/kubectl uncordon worker-1", cleanup.controlCommands.get(0)); + Assert.assertEquals("sudo /opt/bin/kubectl annotate node worker-1 cloudstack.apache.org/cks-live-resize-", cleanup.controlCommands.get(1)); + } + private 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 KubernetesCluster cluster(String uuid) { + KubernetesCluster cluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(cluster.getUuid()).thenReturn(uuid); + return cluster; + } + + private 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; + } + + private KubernetesClusterNodeCapacityReconciler.NodeAccess access() { + return new KubernetesClusterNodeCapacityReconciler.NodeAccess("control", 22, "node", 22, "root", null); + } + + private String nodeJson(boolean unschedulable) { + return String.format("{\"metadata\":{\"annotations\":{}},\"spec\":{\"unschedulable\":%s},\"status\":{\"capacity\":{\"cpu\":\"3900m\",\"memory\":\"4Gi\"},\"allocatable\":{\"cpu\":\"3700m\",\"memory\":\"3900Mi\"},\"conditions\":[{\"type\":\"Ready\",\"status\":\"True\"}]}}", unschedulable); + } + + private static class FakeReconciler extends KubernetesClusterNodeCapacityReconcilerImpl { + private final String nodeJson; + private final List controlCommands = new ArrayList<>(); + + FakeReconciler(String nodeJson) { + this.nodeJson = nodeJson; + } + + @Override + protected String executeControl(KubernetesClusterNodeCapacityReconciler.NodeAccess access, String command) { + controlCommands.add(command); + return command.contains(" get node ") ? nodeJson : ""; + } + + @Override + protected String executeNode(KubernetesClusterNodeCapacityReconciler.NodeAccess access, String command) { + return "2\n2097152\n"; + } + } } From f57838e2374b107a54d2289e04930d6671f035e7 Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 13:37:16 +0200 Subject: [PATCH 03/11] CKS: test live resize reconciliation guards --- .../KubernetesClusterScaleWorkerTest.java | 3 +++ ...etesClusterNodeCapacityReconcilerImplTest.java | 15 +++++++++++++++ 2 files changed, 18 insertions(+) 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 ffd9a28288cc..2c16ea755731 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 @@ -219,5 +219,8 @@ public void testShouldReconcileNodeCapacityOnlyForManagedKvmNodes() { Mockito.when(managedNode.isExternalNode()).thenReturn(false); Mockito.when(kvmNode.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.XenServer); Assert.assertFalse(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); + + Mockito.when(runningManagedCluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged); + Assert.assertFalse(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java index 3707c8014bff..eee5b154888b 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -103,6 +103,21 @@ public void testCordonAndRestoreOnlyNodesOwnedByCks() throws Exception { Assert.assertEquals("sudo /opt/bin/kubectl annotate node worker-1 cloudstack.apache.org/cks-live-resize-", cleanup.controlCommands.get(1)); } + @Test + public void testCordonAddsOwnershipAnnotationBeforeCordoning() throws Exception { + KubernetesCluster cluster = cluster("cluster-uuid"); + UserVm vm = vm("worker-1"); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot schedulable = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, 0, 0, 0, 0, 0, 0, true); + FakeReconciler fakeReconciler = new FakeReconciler(nodeJson(true)); + + fakeReconciler.cordonIfNeeded(cluster, vm, schedulable, access(), System.currentTimeMillis() + 1000L); + + Assert.assertEquals("sudo /opt/bin/kubectl annotate node worker-1 cloudstack.apache.org/cks-live-resize=cluster-uuid --overwrite", + fakeReconciler.controlCommands.get(0)); + Assert.assertEquals("sudo /opt/bin/kubectl cordon worker-1", fakeReconciler.controlCommands.get(1)); + } + private ServiceOffering offering(int cpu, int memory) { ServiceOffering offering = Mockito.mock(ServiceOffering.class); Mockito.when(offering.getCpu()).thenReturn(cpu); From a40ea44fd3a81f43eef36763c1f1bf13a54033c0 Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 13:39:38 +0200 Subject: [PATCH 04/11] CKS: improve live resize failure diagnostics --- .../actionworkers/KubernetesClusterScaleWorker.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) 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 2df102531c66..a7a80dd40b65 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 @@ -440,7 +440,14 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, logger.warn("Unable to restore schedulability for CKS node {} after a pre-resize failure", userVM.getUuid(), cleanupException); } } - 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); + String recovery = resizeSucceeded + ? "The node remains cordoned; check kubelet on the VM and retry the CKS scale operation." + : "The VM was not resized; CKS attempted to restore its temporary cordon."; + String message = String.format("Scaling Kubernetes cluster %s (UUID %s) failed for %s node VM %s (UUID %s), " + + "offering %s to %s: %s %s", kubernetesCluster.getName(), kubernetesCluster.getUuid(), nodeType, + userVM.getDisplayName(), userVM.getUuid(), oldOffering == null ? null : oldOffering.getId(), + serviceOffering.getId(), e.getMessage(), recovery); + logTransitStateAndThrow(Level.ERROR, message, kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e); } 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); From 780d621dbe1e2b57ffa6118a548766b783724d4b Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 13:42:01 +0200 Subject: [PATCH 05/11] CKS: resolve live resize target public IP --- .../cluster/actionworkers/KubernetesClusterScaleWorker.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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 a7a80dd40b65..797e82a7bbc3 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 @@ -475,7 +475,11 @@ private NodeAccess resolveNodeAccess(UserVm userVM) { if (sshRule == null) { throw new CloudRuntimeException(String.format("Unable to resolve SSH port-forwarding rule for VM %s", userVM.getUuid())); } - return new NodeAccess(controlAccess.first(), controlAccess.second(), controlAccess.first(), sshRule.getSourcePortStart(), + IpAddress targetPublicIp = network.getVpcId() == null ? getNetworkSourceNatIp(network) : getVpcTierKubernetesPublicIp(network); + if (targetPublicIp == null) { + throw new CloudRuntimeException(String.format("Unable to resolve target-node public IP for VM %s", userVM.getUuid())); + } + return new NodeAccess(controlAccess.first(), controlAccess.second(), targetPublicIp.getAddress().addr(), sshRule.getSourcePortStart(), getControlNodeLoginUser(), getManagementServerSshPublicKeyFile()); } From 4216946a4741310b381cc283a5adc3b8432383bc Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 22 Sep 2026 17:08:33 +0200 Subject: [PATCH 06/11] CKS: register live resize capacity reconciler --- .../KubernetesClusterNodeCapacityReconcilerImpl.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java index 68576ce12c9e..2c8d9efc4caf 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java @@ -32,8 +32,15 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import org.apache.commons.lang3.StringUtils; - -/** SSH-backed implementation for a rolling CKS live-resize reconciliation. */ +import org.springframework.stereotype.Component; + +/** + * SSH-backed implementation for a rolling CKS live-resize reconciliation. + * + * Component scanning makes this implementation available to the scale worker, which obtains it + * through {@code ComponentContext}. + */ +@Component public class KubernetesClusterNodeCapacityReconcilerImpl implements KubernetesClusterNodeCapacityReconciler { static final long MINIMUM_MEMORY_OVERHEAD_BYTES = 128L * 1024L * 1024L; static final long KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES = 16L * 1024L * 1024L; From 7e5a95a7593f381b1b684e905f417f95465e8210 Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 20 Oct 2026 07:06:35 +0200 Subject: [PATCH 07/11] CKS: retry capacity reconciliation after live resize --- .../KubernetesClusterScaleWorker.java | 27 +++++++++++++------ ...etesClusterNodeCapacityReconcilerImpl.java | 16 +++++------ .../spring-kubernetes-service-context.xml | 2 -- .../KubernetesClusterScaleWorkerTest.java | 26 ++++++++++++++++++ ...ClusterNodeCapacityReconcilerImplTest.java | 17 +++++++++++- 5 files changed, 67 insertions(+), 21 deletions(-) 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 797e82a7bbc3..780f361dbdf0 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 @@ -28,8 +28,6 @@ import java.util.Set; import java.util.stream.Collectors; -import javax.inject.Inject; - import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; import com.cloud.service.ServiceOfferingVO; import com.cloud.storage.VMTemplateVO; @@ -87,8 +85,7 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif private Boolean isAutoscalingEnabled; private long scaleTimeoutTime; - @Inject - protected KubernetesClusterNodeCapacityReconciler kubernetesClusterNodeCapacityReconciler; + protected KubernetesClusterNodeCapacityReconciler kubernetesClusterNodeCapacityReconciler = new KubernetesClusterNodeCapacityReconcilerImpl(); protected KubernetesClusterScaleWorker(final KubernetesCluster kubernetesCluster, final KubernetesClusterManagerImpl clusterManager) { super(kubernetesCluster, clusterManager); @@ -401,26 +398,30 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); } boolean reconcileKubernetes = shouldReconcileNodeCapacity(vmMapVO, userVM, oldOffering, serviceOffering, nodeType); + boolean reconciliationRetryCandidate = isReconciliationRetryCandidate(vmMapVO, userVM, serviceOffering, nodeType); NodeAccess nodeAccess = null; NodeCapacitySnapshot before = null; boolean result = false; boolean cksCordonCompleted = false; - boolean resizeSucceeded = serviceOffering.getId() == userVM.getServiceOfferingId(); + boolean resizeSucceeded = Objects.equals(serviceOffering.getId(), userVM.getServiceOfferingId()); try { - if (reconcileKubernetes) { + if (reconcileKubernetes || reconciliationRetryCandidate) { nodeAccess = resolveNodeAccess(userVM); before = kubernetesClusterNodeCapacityReconciler.captureBefore(kubernetesCluster, userVM, nodeAccess); + reconcileKubernetes = reconcileKubernetes || before.isCloudStackResizeCordon(); + } + if (reconcileKubernetes) { kubernetesClusterNodeCapacityReconciler.cordonIfNeeded(kubernetesCluster, userVM, before, nodeAccess, scaleTimeoutTime); cksCordonCompleted = !before.isUnschedulable() || before.isCloudStackResizeCordon(); } - if (serviceOffering.getId() != userVM.getServiceOfferingId()) { + if (!Objects.equals(serviceOffering.getId(), userVM.getServiceOfferingId())) { result = userVmManager.upgradeVirtualMachine(userVM.getId(), serviceOffering.getId(), new HashMap()); 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); } resizeSucceeded = true; userVM = userVmDao.findById(userVM.getId()); - if (userVM == null || serviceOffering.getId() != userVM.getServiceOfferingId()) { + if (userVM == null || !Objects.equals(serviceOffering.getId(), userVM.getServiceOfferingId())) { logTransitStateAndThrow(Level.WARN, String.format("Scaling Kubernetes cluster : %s failed, VM %s did not reach target offering", kubernetesCluster.getName(), vmMapVO.getVmId()),kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed); } } @@ -492,6 +493,16 @@ protected boolean shouldReconcileNodeCapacity(KubernetesClusterVmMapVO vmMapVO, && kubernetesClusterNodeCapacityReconciler.requiresKubeletRefresh(oldOffering, targetOffering, nodeType, originalState); } + protected boolean isReconciliationRetryCandidate(KubernetesClusterVmMapVO vmMapVO, UserVm userVM, + ServiceOffering targetOffering, KubernetesClusterNodeType nodeType) { + return !vmMapVO.isExternalNode() + && KubernetesCluster.ClusterType.CloudManaged == kubernetesCluster.getClusterType() + && Hypervisor.HypervisorType.KVM == userVM.getHypervisorType() + && (WORKER == nodeType || CONTROL == nodeType) + && KubernetesCluster.State.Running == originalState + && Objects.equals(targetOffering.getId(), userVM.getServiceOfferingId()); + } + private void removeNodesFromCluster(List vmMaps) throws CloudRuntimeException { for (KubernetesClusterVmMapVO vmMapVO : vmMaps) { UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId()); diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java index 2c8d9efc4caf..2502490238d5 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java @@ -32,17 +32,13 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import org.apache.commons.lang3.StringUtils; -import org.springframework.stereotype.Component; - -/** - * SSH-backed implementation for a rolling CKS live-resize reconciliation. - * - * Component scanning makes this implementation available to the scale worker, which obtains it - * through {@code ComponentContext}. - */ -@Component + +/** SSH-backed implementation for a rolling CKS live-resize reconciliation. */ public class KubernetesClusterNodeCapacityReconcilerImpl implements KubernetesClusterNodeCapacityReconciler { - static final long MINIMUM_MEMORY_OVERHEAD_BYTES = 128L * 1024L * 1024L; + // KVM QA observes 4,568,788 KiB in a 4,608 MiB guest after memory hotplug + // (about 146 MiB below the offering), so keep a small margin above that + // platform overhead while still rejecting materially undersized guests. + static final long MINIMUM_MEMORY_OVERHEAD_BYTES = 160L * 1024L * 1024L; static final long KUBERNETES_MEMORY_REPORTING_TOLERANCE_BYTES = 16L * 1024L * 1024L; private static final int COMMAND_TIMEOUT_MS = 30000; private static final int POLL_INTERVAL_MS = 5000; diff --git a/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml b/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml index 9fe19200d6e7..fedc57071c5d 100644 --- a/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml +++ b/plugins/integrations/kubernetes-service/src/main/resources/META-INF/cloudstack/kubernetes-service/spring-kubernetes-service-context.xml @@ -34,8 +34,6 @@ - - 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 2c16ea755731..8a35c06f0c3f 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 @@ -21,6 +21,7 @@ import com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl; import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao; import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconciler; +import com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconcilerImpl; import com.cloud.hypervisor.Hypervisor; import com.cloud.offering.ServiceOffering; import com.cloud.service.ServiceOfferingVO; @@ -193,6 +194,11 @@ public void testGetWorkerNodesToRemoveForDownsize_noRemoval() { Assert.assertTrue(toRemove.isEmpty()); } + @Test + public void testUsesLiveResizeReconcilerWithoutComponentContextRegistration() { + Assert.assertTrue(worker.kubernetesClusterNodeCapacityReconciler instanceof KubernetesClusterNodeCapacityReconcilerImpl); + } + @Test public void testShouldReconcileNodeCapacityOnlyForManagedKvmNodes() { KubernetesCluster runningManagedCluster = Mockito.mock(KubernetesCluster.class); @@ -223,4 +229,24 @@ public void testShouldReconcileNodeCapacityOnlyForManagedKvmNodes() { Mockito.when(runningManagedCluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged); Assert.assertFalse(scaleWorker.shouldReconcileNodeCapacity(managedNode, kvmNode, oldOffering, targetOffering, WORKER)); } + + @Test + public void testCksOwnedCordonCanRetryReconciliationAfterVmReachedTargetOffering() { + KubernetesCluster runningManagedCluster = Mockito.mock(KubernetesCluster.class); + Mockito.when(runningManagedCluster.getState()).thenReturn(KubernetesCluster.State.Running); + Mockito.when(runningManagedCluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.CloudManaged); + KubernetesClusterScaleWorker scaleWorker = new KubernetesClusterScaleWorker(runningManagedCluster, + new java.util.HashMap<>(), 1L, null, false, null, null, clusterManager); + + KubernetesClusterVmMapVO managedNode = Mockito.mock(KubernetesClusterVmMapVO.class); + Mockito.when(managedNode.isExternalNode()).thenReturn(false); + UserVmVO kvmNode = Mockito.mock(UserVmVO.class); + Mockito.when(kvmNode.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM); + Mockito.when(kvmNode.getServiceOfferingId()).thenReturn(Long.parseLong("1024")); + ServiceOffering targetOffering = Mockito.mock(ServiceOffering.class); + Mockito.when(targetOffering.getId()).thenReturn(Long.parseLong("1024")); + + Assert.assertTrue(scaleWorker.isReconciliationRetryCandidate(managedNode, kvmNode, targetOffering, WORKER)); + Assert.assertFalse(scaleWorker.isReconciliationRetryCandidate(managedNode, kvmNode, targetOffering, DEFAULT)); + } } diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java index eee5b154888b..f54d9e566e59 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -48,6 +48,8 @@ public void testOnlyRunningKubernetesNodesWithChangedCpuOrRamRequireRefresh() { Assert.assertTrue(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); Assert.assertTrue(reconciler.requiresKubeletRefresh(oldOffering, ramOffering, KubernetesClusterNodeType.CONTROL, KubernetesCluster.State.Running)); Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, capOnlyOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); + Assert.assertFalse(reconciler.requiresKubeletRefresh(null, cpuOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); + Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, null, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Running)); Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.ETCD, KubernetesCluster.State.Running)); Assert.assertFalse(reconciler.requiresKubeletRefresh(oldOffering, cpuOffering, KubernetesClusterNodeType.WORKER, KubernetesCluster.State.Stopped)); } @@ -118,6 +120,18 @@ public void testCordonAddsOwnershipAnnotationBeforeCordoning() throws Exception Assert.assertEquals("sudo /opt/bin/kubectl cordon worker-1", fakeReconciler.controlCommands.get(1)); } + @Test + public void testGuestVerificationAllowsObservedKvmHotplugOverhead() throws Exception { + FakeReconciler fakeReconciler = new FakeReconciler(nodeJson(false)); + fakeReconciler.nodeResources = "4\n4568788\n"; + + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot before = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, 2, 2097152L, 2000L, 2L * 1024L * 1024L * 1024L, + 1900L, 1900L * 1024L * 1024L, true); + + fakeReconciler.verifyGuestResources(vm("worker-1"), offering(4, 4608), before, access(), System.currentTimeMillis() + 1000L); + } + private ServiceOffering offering(int cpu, int memory) { ServiceOffering offering = Mockito.mock(ServiceOffering.class); Mockito.when(offering.getCpu()).thenReturn(cpu); @@ -149,6 +163,7 @@ private String nodeJson(boolean unschedulable) { private static class FakeReconciler extends KubernetesClusterNodeCapacityReconcilerImpl { private final String nodeJson; private final List controlCommands = new ArrayList<>(); + private String nodeResources = "2\n2097152\n"; FakeReconciler(String nodeJson) { this.nodeJson = nodeJson; @@ -162,7 +177,7 @@ protected String executeControl(KubernetesClusterNodeCapacityReconciler.NodeAcce @Override protected String executeNode(KubernetesClusterNodeCapacityReconciler.NodeAccess access, String command) { - return "2\n2097152\n"; + return nodeResources; } } } From 5480119108462433a2b611b7d25d67e2e92f627c Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 20 Oct 2026 14:59:56 +0200 Subject: [PATCH 08/11] CKS: online guest resources after live resize --- .../actionworkers/KubernetesClusterScaleWorker.java | 1 + .../KubernetesClusterNodeCapacityReconciler.java | 1 + .../KubernetesClusterNodeCapacityReconcilerImpl.java | 9 +++++++++ ...ernetesClusterNodeCapacityReconcilerImplTest.java | 12 ++++++++++++ 4 files changed, 23 insertions(+) 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 780f361dbdf0..2ce1332d5c31 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 @@ -426,6 +426,7 @@ private void scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType, } } if (reconcileKubernetes) { + kubernetesClusterNodeCapacityReconciler.onlineGuestResources(nodeAccess); kubernetesClusterNodeCapacityReconciler.verifyGuestResources(userVM, serviceOffering, before, nodeAccess, scaleTimeoutTime); if (!kubernetesClusterNodeCapacityReconciler.isKubernetesResourcesCurrent(before, serviceOffering)) { kubernetesClusterNodeCapacityReconciler.restartKubelet(userVM, nodeAccess, scaleTimeoutTime); 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 index f52458774aca..3fbabca6e322 100644 --- 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 @@ -29,6 +29,7 @@ public interface KubernetesClusterNodeCapacityReconciler { boolean requiresKubeletRefresh(ServiceOffering oldOffering, ServiceOffering newOffering, KubernetesClusterNodeType nodeType, KubernetesCluster.State originalState); void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; + void onlineGuestResources(NodeAccess access) throws Exception; void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception; boolean isKubernetesResourcesCurrent(NodeCapacitySnapshot snapshot, ServiceOffering target); void restartKubelet(UserVm vm, NodeAccess access, long deadline) throws Exception; diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java index 2502490238d5..bec9b91a83e9 100644 --- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java +++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java @@ -45,6 +45,10 @@ public class KubernetesClusterNodeCapacityReconcilerImpl implements KubernetesCl private static final Pattern NODE_NAME = Pattern.compile("[A-Za-z0-9][A-Za-z0-9.-]{0,252}"); private static final Pattern MEMORY_QUANTITY = Pattern.compile("([0-9]+)(Ki|Mi|Gi|Ti|K|M|G|T)?"); private static final String RESIZE_ANNOTATION = "cloudstack.apache.org/cks-live-resize"; + 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'"; @Override public boolean requiresKubeletRefresh(ServiceOffering oldOffering, ServiceOffering newOffering, @@ -72,6 +76,11 @@ public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySna throw failure("CORDON", vm, "Kubernetes node did not become unschedulable"); } + @Override + public void onlineGuestResources(NodeAccess access) throws Exception { + executeNode(access, ONLINE_HOTPLUGGED_RESOURCES); + } + @Override public void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) throws Exception { while (System.currentTimeMillis() < deadline) { diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java index f54d9e566e59..7611f582f13b 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -132,6 +132,16 @@ public void testGuestVerificationAllowsObservedKvmHotplugOverhead() throws Excep fakeReconciler.verifyGuestResources(vm("worker-1"), offering(4, 4608), before, access(), System.currentTimeMillis() + 1000L); } + @Test + public void testOnlineGuestResourcesActivatesHotpluggedCpuAndMemory() throws Exception { + FakeReconciler fakeReconciler = new FakeReconciler(nodeJson(false)); + + fakeReconciler.onlineGuestResources(access()); + + Assert.assertEquals(KubernetesClusterNodeCapacityReconcilerImpl.ONLINE_HOTPLUGGED_RESOURCES, + fakeReconciler.nodeCommands.get(0)); + } + private ServiceOffering offering(int cpu, int memory) { ServiceOffering offering = Mockito.mock(ServiceOffering.class); Mockito.when(offering.getCpu()).thenReturn(cpu); @@ -163,6 +173,7 @@ private String nodeJson(boolean unschedulable) { private static class FakeReconciler extends KubernetesClusterNodeCapacityReconcilerImpl { private final String nodeJson; private final List controlCommands = new ArrayList<>(); + private final List nodeCommands = new ArrayList<>(); private String nodeResources = "2\n2097152\n"; FakeReconciler(String nodeJson) { @@ -177,6 +188,7 @@ protected String executeControl(KubernetesClusterNodeCapacityReconciler.NodeAcce @Override protected String executeNode(KubernetesClusterNodeCapacityReconciler.NodeAccess access, String command) { + nodeCommands.add(command); return nodeResources; } } From 2a404aa1ffcccc2cc4b1e131562856c1fc7b754c Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 20 Oct 2026 15:36:50 +0200 Subject: [PATCH 09/11] CKS: recompute cluster capacity from node type offerings --- .../KubernetesClusterScaleWorker.java | 59 +++++++-------- .../KubernetesClusterScaleWorkerTest.java | 73 ++++++++++++++----- 2 files changed, 81 insertions(+), 51 deletions(-) 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 2ce1332d5c31..4d8852611ca6 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 @@ -29,7 +29,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; @@ -219,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; + serviceOfferingDao.findById(getConfiguredOfferingIdForNodeType(nodeType)) : newServiceOffering; final Long serviceOfferingId = newServiceOffering == null ? null : serviceOffering.getId(); Pair clusterCountAndCapacity = calculateNewClusterCountAndCapacity(newWorkerSize, nodeType, serviceOffering); @@ -236,32 +235,36 @@ 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 : List.of(CONTROL, WORKER, ETCD)) { + 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 : + serviceOfferingDao.findByIdIncludingRemoved(getConfiguredOfferingIdForNodeType(type)); + Pair capacity = calculateNodesCapacity(offering, nodeCount); + cores += capacity.first(); + memory += capacity.second(); } return new Pair<>(cores, memory); } + private Long getConfiguredOfferingIdForNodeType(KubernetesClusterNodeType nodeType) { + Long offeringId = null; + if (WORKER == nodeType) { + offeringId = kubernetesCluster.getWorkerNodeServiceOfferingId(); + } else if (CONTROL == nodeType) { + offeringId = kubernetesCluster.getControlNodeServiceOfferingId(); + } else if (ETCD == nodeType) { + offeringId = kubernetesCluster.getEtcdNodeServiceOfferingId(); + } + return offeringId == null ? kubernetesCluster.getServiceOfferingId() : offeringId; + } + private long getNodeCountForType(KubernetesClusterNodeType nodeType, KubernetesCluster kubernetesCluster) { if (WORKER == nodeType) { return kubernetesCluster.getNodeCount(); @@ -273,16 +276,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); } 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 8a35c06f0c3f..170035d71355 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 @@ -98,21 +98,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; @@ -120,12 +112,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); @@ -134,6 +120,57 @@ 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 testGetWorkerNodesToRemoveForDownsize_singleRemoval() { From 978b29173ed70c695b50bce9ea7ecdf7436ae070 Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Tue, 20 Oct 2026 20:16:48 +0200 Subject: [PATCH 10/11] CKS: preserve node offerings across scale and retry paths --- .../KubernetesClusterScaleWorker.java | 20 +++--- .../KubernetesClusterScaleWorkerTest.java | 65 +++++++++++++++++++ 2 files changed, 73 insertions(+), 12 deletions(-) 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 4d8852611ca6..0b53c89b6183 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 @@ -338,7 +338,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 +361,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); @@ -708,6 +708,9 @@ public boolean scaleCluster() throws CloudRuntimeException { for (KubernetesClusterNodeType nodeType : Arrays.asList(CONTROL, ETCD, WORKER)) { boolean isWorkerNode = WORKER == nodeType; final long newVMRequired = (!isWorkerNode || clusterSize == null) ? 0 : clusterSize - originalClusterSize; + if (nodeType == ETCD && kubernetesCluster.getEtcdNodeCount() == 0) { + continue; + } if (!scaleClusterDefaultOffering && !serviceOfferingNodeTypeMap.containsKey(nodeType.name()) && newVMRequired == 0) { continue; } @@ -777,15 +780,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 getConfiguredOfferingIdForNodeType(nodeType); } } 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 170035d71355..ac4ece5a268a 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 @@ -43,6 +43,7 @@ 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) @@ -171,6 +172,70 @@ private ServiceOfferingVO offering(int cpu, int 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() { From f92e52d736b9491309f6be3baac250de18c198c9 Mon Sep 17 00:00:00 2001 From: andrijapanicsb Date: Wed, 21 Oct 2026 17:31:12 +0200 Subject: [PATCH 11/11] test(cks): cover live resize capacity reconciliation failure paths --- ...ClusterNodeCapacityReconcilerImplTest.java | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) diff --git a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java index 7611f582f13b..0e37dd8a5839 100644 --- a/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java +++ b/plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImplTest.java @@ -23,6 +23,7 @@ import com.cloud.kubernetes.cluster.KubernetesServiceHelper.KubernetesClusterNodeType; import com.cloud.offering.ServiceOffering; import com.cloud.uservm.UserVm; +import com.cloud.utils.exception.CloudRuntimeException; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -142,6 +143,175 @@ public void testOnlineGuestResourcesActivatesHotpluggedCpuAndMemory() throws Exc fakeReconciler.nodeCommands.get(0)); } + @Test + public void testParseMemoryDecimalBinaryAndUnitlessQuantities() { + Assert.assertEquals(123L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("123")); + Assert.assertEquals(2L * 1024L * 1024L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2Mi")); + Assert.assertEquals(2L * 1024L * 1024L * 1024L * 1024L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2Ti")); + Assert.assertEquals(2000L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2K")); + Assert.assertEquals(2000000L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2M")); + Assert.assertEquals(2000000000L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2G")); + Assert.assertEquals(2000000000000L, KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("2T")); + Assert.assertThrows(IllegalArgumentException.class, () -> KubernetesClusterNodeCapacityReconcilerImpl.parseMemoryBytes("4Gi; reboot")); + } + + @Test + public void testRestartKubeletWaitsForActiveService() throws Exception { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + fake.nodeResources = " active\n"; + + fake.restartKubelet(vm("worker-1"), access(), System.currentTimeMillis() + 1000L); + + Assert.assertEquals(List.of("sudo systemctl restart kubelet", "sudo systemctl is-active kubelet"), fake.nodeCommands); + } + + @Test + public void testExpiredDeadlinesReportTheFailedPhaseAndVm() { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + KubernetesCluster cluster = cluster("cluster-uuid"); + UserVm vm = vm("worker-1"); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot before = snapshot(2, 2048, true); + long expired = System.currentTimeMillis() - 1L; + + assertPhaseFailure("CORDON", () -> fake.cordonIfNeeded(cluster, vm, before, access(), expired)); + assertPhaseFailure("GUEST_VERIFY", () -> fake.verifyGuestResources(vm, offering(4, 4096), before, access(), expired)); + assertPhaseFailure("KUBELET_RESTART", () -> fake.restartKubelet(vm, access(), expired)); + assertPhaseFailure("CAPACITY_VERIFY", () -> fake.waitForKubernetesResources(cluster, vm, offering(4, 4096), before, access(), expired)); + assertPhaseFailure("RESTORE_SCHEDULABILITY", () -> fake.restoreSchedulability(cluster, vm, before, access(), expired)); + } + + @Test + public void testWaitForKubernetesResourcesRequiresIncreasedAllocatableCpuAndMemory() throws Exception { + FakeReconciler fake = new FakeReconciler(nodeJson(false).replace("3900m", "4")); + fake.nodeResources = "4\n4194304\n"; + + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot observed = fake.waitForKubernetesResources( + cluster("cluster-uuid"), vm("worker-1"), offering(4, 4096), snapshot(2, 2048, true), + access(), System.currentTimeMillis() + 1000L); + + Assert.assertTrue(observed.isReady()); + Assert.assertEquals(4000L, observed.getCapacityCpuMillis()); + Assert.assertEquals(3700L, observed.getAllocatableCpuMillis()); + Assert.assertEquals(3900L * 1024L * 1024L, observed.getAllocatableMemoryBytes()); + } + + @Test(timeout = 15000L) + public void testWaitRejectsUnchangedAllocatableCpuAfterCpuIncrease() { + FakeReconciler fake = new FakeReconciler(nodeJson(false).replace("3900m", "4").replace("3700m", "1900m")); + fake.nodeResources = "4\n4194304\n"; + + assertPhaseFailure("CAPACITY_VERIFY", () -> fake.waitForKubernetesResources(cluster("cluster-uuid"), vm("worker-1"), + offering(4, 4096), snapshot(2, 2048, true), access(), System.currentTimeMillis() + 1000L)); + + Assert.assertFalse(fake.controlCommands.isEmpty()); + Assert.assertTrue(fake.controlCommands.stream().noneMatch(command -> command.contains("uncordon"))); + } + + @Test(timeout = 15000L) + public void testWaitRejectsStaleAllocatableMemoryAfterMemoryIncrease() { + FakeReconciler fake = new FakeReconciler(nodeJson(false).replace("3900m", "4").replace("3900Mi", "1900Mi")); + fake.nodeResources = "4\n4194304\n"; + + assertPhaseFailure("CAPACITY_VERIFY", () -> fake.waitForKubernetesResources(cluster("cluster-uuid"), vm("worker-1"), + offering(4, 4096), snapshot(2, 2048, true), access(), System.currentTimeMillis() + 1000L)); + + Assert.assertFalse(fake.controlCommands.isEmpty()); + } + + @Test(timeout = 15000L) + public void testRestartKubeletDoesNotTreatInactiveServiceAsSuccess() { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + fake.nodeResources = "inactive\n"; + + assertPhaseFailure("KUBELET_RESTART", () -> fake.restartKubelet(vm("worker-1"), access(), System.currentTimeMillis() + 1000L)); + + Assert.assertEquals(List.of("sudo systemctl restart kubelet", "sudo systemctl is-active kubelet"), fake.nodeCommands); + } + + @Test + public void testResourceVerificationRejectsStaleNotReadyOrOutOfToleranceNodes() { + ServiceOffering target = offering(4, 4096); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(snapshot(4, 4096, false), target)); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(snapshot(2, 4096, true), target)); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(snapshot(4, 4097, true), target)); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(snapshot(4, 3800, true), target)); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot staleMemory = + new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, 4, 4194304L, + 4000L, 2L * 1024L * 1024L * 1024L, 3700L, 1900L * 1024L * 1024L, true); + Assert.assertFalse(reconciler.isKubernetesResourcesCurrent(staleMemory, target)); + } + + @Test + public void testCaptureRecognizesOnlyThisClustersCordonOwnership() throws Exception { + String owned = nodeJson(true).replace("\"annotations\":{}", + "\"annotations\":{\"cloudstack.apache.org/cks-live-resize\":\"cluster-uuid\"}"); + FakeReconciler fake = new FakeReconciler(owned); + Assert.assertTrue(fake.captureBefore(cluster("cluster-uuid"), vm("worker-1"), access()).isCloudStackResizeCordon()); + Assert.assertFalse(fake.captureBefore(cluster("another-cluster"), vm("worker-1"), access()).isCloudStackResizeCordon()); + } + + @Test + public void testCaptureHandlesMissingOptionalFieldsAndNotReadyNode() throws Exception { + String minimal = nodeJson(false).replace("\"annotations\":{}", "\"name\":\"worker-1\"") + .replace("\"spec\":{\"unschedulable\":false},", "") + .replace("\"status\":\"True\"", "\"status\":\"False\""); + FakeReconciler fake = new FakeReconciler(minimal); + KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot observed = + fake.captureBefore(cluster("cluster-uuid"), vm("worker-1"), access()); + Assert.assertFalse(observed.isReady()); + Assert.assertFalse(observed.isUnschedulable()); + Assert.assertFalse(observed.isCloudStackResizeCordon()); + } + + @Test + public void testInvalidNodeNameIsRejectedBeforeControlCommand() { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + Assert.assertThrows(CloudRuntimeException.class, + () -> fake.captureBefore(cluster("cluster-uuid"), vm("worker; reboot"), access())); + Assert.assertTrue(fake.controlCommands.isEmpty()); + } + + @Test + public void testNodeNameIsLowercaseInKubectlCommand() throws Exception { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + fake.captureBefore(cluster("cluster-uuid"), vm("Worker-1"), access()); + Assert.assertEquals("sudo /opt/bin/kubectl get node worker-1 -o json", fake.controlCommands.get(0)); + } + + @Test + public void testMalformedGuestResourceOutputStopsBeforeKubectl() { + FakeReconciler fake = new FakeReconciler(nodeJson(false)); + fake.nodeResources = "4\n"; + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, + () -> fake.captureBefore(cluster("cluster-uuid"), vm("worker-1"), access())); + Assert.assertEquals("Unable to read guest CPU and memory", error.getMessage()); + Assert.assertTrue(fake.controlCommands.isEmpty()); + } + + @Test + public void testNodeAccessPreservesSeparateControlAndWorkerEndpoints() { + java.io.File key = new java.io.File("test-only-key"); + KubernetesClusterNodeCapacityReconciler.NodeAccess access = + new KubernetesClusterNodeCapacityReconciler.NodeAccess("control", 2222, "worker", 2223, "cloud", key); + Assert.assertEquals("control", access.getControlAddress()); + Assert.assertEquals(2222, access.getControlPort()); + Assert.assertEquals("worker", access.getNodeAddress()); + Assert.assertEquals(2223, access.getNodePort()); + Assert.assertEquals("cloud", access.getUser()); + Assert.assertSame(key, access.getSshKeyFile()); + } + + private KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot snapshot(int cpu, int memoryMiB, boolean ready) { + long bytes = memoryMiB * 1024L * 1024L; + return new KubernetesClusterNodeCapacityReconciler.NodeCapacitySnapshot(false, false, cpu, memoryMiB * 1024L, + cpu * 1000L, bytes, cpu * 1000L - 100L, bytes - 128L * 1024L * 1024L, ready); + } + + private void assertPhaseFailure(String phase, org.junit.function.ThrowingRunnable action) { + CloudRuntimeException error = Assert.assertThrows(CloudRuntimeException.class, action); + Assert.assertTrue(error.getMessage(), error.getMessage().contains(phase + " failed for VM vm-uuid")); + } + private ServiceOffering offering(int cpu, int memory) { ServiceOffering offering = Mockito.mock(ServiceOffering.class); Mockito.when(offering.getCpu()).thenReturn(cpu);