You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by an...@apache.org on 2026/09/22 10:47:47 UTC
(cloudstack) 01/01: CKS: reconcile node capacity after live resize
This is an automated email from the ASF dual-hosted git repository.
andrijapanicsb pushed a commit to branch cks-live-resize-capacity-reconciliation
in repository https://gitbox.apache.org/repos/asf/cloudstack.git
commit 6f6656f491717a9df02fee3e0f4951cb4821945c
Author: andrijapanicsb <an...@gmail.com>
AuthorDate: Tue Sep 22 12:47:24 2026 +0200
CKS: reconcile node capacity after live resize
---
.../KubernetesClusterScaleWorker.java | 99 ++++++++++-
.../KubernetesClusterNodeCapacityReconciler.java | 67 ++++++++
...ubernetesClusterNodeCapacityReconcilerImpl.java | 190 +++++++++++++++++++++
.../spring-kubernetes-service-context.xml | 1 +
.../KubernetesClusterScaleWorkerTest.java | 33 ++++
...netesClusterNodeCapacityReconcilerImplTest.java | 71 ++++++++
6 files changed, 455 insertions(+), 6 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 08513dbd448..a1dbf28b2bd 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.Objects;
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;
@@ -43,16 +45,21 @@ import com.cloud.exception.InsufficientCapacityException;
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;
@@ -79,6 +86,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);
}
@@ -359,15 +369,60 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif
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<String, String>());
- } 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<String, String>());
+ 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);
}
@@ -375,6 +430,38 @@ public class KubernetesClusterScaleWorker extends KubernetesClusterResourceModif
kubernetesCluster = updateKubernetesClusterEntryForNodeType(null, nodeType, serviceOffering, updateNodeOffering, updateClusterOffering);
}
+ private NodeAccess resolveNodeAccess(UserVm userVM) {
+ Pair<String, Integer> 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<KubernetesClusterVmMapVO> 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 00000000000..f52458774ac
--- /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 00000000000..a6785ec3fdf
--- /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<Boolean, String> 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 05336678629..9fe19200d6e 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 @@
<bean id="kubernetesClusterVmMapDaoImpl" class="com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDaoImpl" />
<bean id="kubernetesClusterAffinityGroupMapDaoImpl" class="com.cloud.kubernetes.cluster.dao.KubernetesClusterAffinityGroupMapDaoImpl" />
<bean id="kubernetesClusterManagerImpl" class="com.cloud.kubernetes.cluster.KubernetesClusterManagerImpl" />
+ <bean id="kubernetesClusterNodeCapacityReconciler" class="com.cloud.kubernetes.cluster.utils.KubernetesClusterNodeCapacityReconcilerImpl" />
<bean id="kubernetesServiceHelper" class="com.cloud.kubernetes.cluster.KubernetesServiceHelperImpl" >
<property name="name" value="KubernetesServiceHelper" />
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 c9299bdbaa6..ffd9a28288c 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.KubernetesCluster;
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 java.util.List;
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 class KubernetesClusterScaleWorkerTest {
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 00000000000..729803c7187
--- /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;
+ }
+}