andrijapanicsb commented on code in PR #14223: URL: https://github.com/apache/cloudstack/pull/14223#discussion_r4394974429
########## plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java: ########## @@ -0,0 +1,232 @@ +// 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 { + // 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; + 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, + 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) { + return nodeSnapshot(cluster, vm, access, guestResources(access)); + } + + @Override + public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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) { Review Comment: Done, there's one polling helper now, and each caller still reports its own phase in the error. ########## plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java: ########## @@ -0,0 +1,232 @@ +// 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 { + // 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; + 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, + 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) { + return nodeSnapshot(cluster, vm, access, guestResources(access)); + } + + @Override + public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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"); Review Comment: I looked at this one closely. The resize needs the annotations and the cordon/uncordon to change in a single update, so a failure can't leave a node half-marked. The existing helper also loses the thread's interrupt flag. So instead of reusing it as is, I added a small helper next to it in `KubernetesClusterUtil` that builds the node patch command, and moved the node-name handling (including the lowercasing) there, so both use it. ########## plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java: ########## @@ -0,0 +1,232 @@ +// 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 { + // 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; + 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, + 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) { + return nodeSnapshot(cluster, vm, access, guestResources(access)); + } + + @Override + public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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 onlineGuestResources(NodeAccess access) { + executeNode(access, ONLINE_HOTPLUGGED_RESOURCES); + } + + @Override + public void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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); Review Comment: Good catch, the "grew" part meant nothing there, so I removed it. Kubelet computes allocatable as capacity minus the reservations in the same node-status update (the `MachineInfo` setter in `pkg/kubelet/nodestatus/setters.go`), so the check now uses a fresh snapshot: capacity matches the guest, the node is Ready, and allocatable is between 0 and capacity. Kubelet also has to be active before a restart is skipped. Renamed to `isKubernetesNodeAlreadyResized` as you suggested. ########## plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconcilerImpl.java: ########## @@ -0,0 +1,232 @@ +// 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 { + // 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; + 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, + 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) { + return nodeSnapshot(cluster, vm, access, guestResources(access)); + } + + @Override + public void cordonIfNeeded(KubernetesCluster cluster, UserVm vm, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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 onlineGuestResources(NodeAccess access) { + executeNode(access, ONLINE_HOTPLUGGED_RESOURCES); + } + + @Override + public void verifyGuestResources(UserVm vm, ServiceOffering target, NodeCapacitySnapshot before, NodeAccess access, long deadline) { + 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) { + 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) { + 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) { + 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"); Review Comment: Done, one case per line, using the `ByteScaleUtils` constants for the binary units (TiB is derived from them, since there's no constant for it). The decimal units K, M, G, T stay as powers of 1000. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
