CKS: reconcile node capacity after live resize

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java

@@ -976,6 +976,14 @@ public void setKeys(String[] keys) {
 
     protected ServiceOffering getServiceOfferingForNodeTypeOnCluster(KubernetesClusterNodeType nodeType,
                                                                      KubernetesCluster cluster) {
+        return serviceOfferingDao.findById(getServiceOfferingIdForNodeTypeOnCluster(nodeType, cluster));
+    }
+
+    /**
+     * The offering ID set for a node type on the cluster, falling back to the cluster offering.
+     */
+    protected Long getServiceOfferingIdForNodeTypeOnCluster(KubernetesClusterNodeType nodeType,
+                                                            KubernetesCluster cluster) {
         Long offeringId = null;
         Long defaultOfferingId = cluster.getServiceOfferingId();
         Long controlOfferingId = cluster.getControlNodeServiceOfferingId();
@@ -994,7 +1002,7 @@ protected ServiceOffering getServiceOfferingForNodeTypeOnCluster(KubernetesClust
             logger.error(msg);
             throw new CloudRuntimeException(msg);
         }
-        return serviceOfferingDao.findById(offeringId);
+        return offeringId;
     }
 
     protected boolean isDefaultTemplateUsed() {

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/GuestResources.java

@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.kubernetes.cluster.utils;
+
+import org.apache.cloudstack.utils.bytescale.ByteScaleUtils;
+
+/** CPU and memory as seen inside a node VM. */
+public final class GuestResources {
+    private final long onlineCpus;
+    private final long memoryKiB;
+    private final long onlineMemoryBytes;
+    private final long memoryBlockBytes;
+
+    /**
+     * @param onlineCpus        CPUs online in the guest
+     * @param memoryKiB         MemTotal from /proc/meminfo, which kubelet reports as the node's memory capacity
+     * @param onlineMemoryBytes size of the online memory blocks under /sys/devices/system/memory
+     * @param memoryBlockBytes  size of one of those memory blocks
+     */
+    public GuestResources(long onlineCpus,
+                          long memoryKiB,
+                          long onlineMemoryBytes,
+                          long memoryBlockBytes) {
+        this.onlineCpus = onlineCpus;
+        this.memoryKiB = memoryKiB;
+        this.onlineMemoryBytes = onlineMemoryBytes;
+        this.memoryBlockBytes = memoryBlockBytes;
+    }
+
+    public long getOnlineCpus() {
+        return onlineCpus;
+    }
+
+    public long getMemoryKiB() {
+        return memoryKiB;
+    }
+
+    public long getMemoryBytes() {
+        return memoryKiB * ByteScaleUtils.KiB;
+    }
+
+    public long getOnlineMemoryBytes() {
+        return onlineMemoryBytes;
+    }
+
+    public long getMemoryBlockBytes() {
+        return memoryBlockBytes;
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterUtil.java

@@ -50,6 +50,25 @@ public class KubernetesClusterUtil {
 
     public static final String CLUSTER_NODE_READY_COMMAND = "sudo /opt/bin/kubectl get nodes | awk '{if ($1 == \"%s\" && $2 == \"Ready\") print $1}'";
     public static final String CLUSTER_NODE_VERSION_COMMAND = "sudo /opt/bin/kubectl get nodes | awk '{if ($1 == \"%s\") print $5}'";
+    public static final String CLUSTER_NODE_PATCH_COMMAND = "sudo /opt/bin/kubectl patch node %s --type merge --request-timeout=20s -p '%s'";
+
+    /**
+     * The Kubernetes node name of a CKS cluster VM, which is its host name in lowercase.
+     */
+    public static String getKubernetesNodeName(final UserVm userVm) {
+        return StringUtils.lowerCase(userVm.getHostName());
+    }
+
+    /**
+     * Command that applies a JSON merge patch to a node. Annotations and spec.unschedulable in one patch
+     * change in a single API update, so a failure can't leave the node with only some of them changed.
+     */
+    public static String getKubernetesNodePatchCommand(final String nodeName, final String mergePatch) {
+        if (StringUtils.contains(mergePatch, '\'')) {
+            throw new IllegalArgumentException("A node patch must not contain single quotes");
+        }
+        return String.format(CLUSTER_NODE_PATCH_COMMAND, nodeName, mergePatch);
+    }
 
     public static boolean isKubernetesClusterNodeReady(final KubernetesCluster kubernetesCluster, String ipAddress, int port,
                                                        String user, File sshKeyFile, String nodeName) throws Exception {
@@ -108,10 +127,7 @@ public static boolean uncordonKubernetesClusterNode(final KubernetesCluster kube
                                                         final String user, final File sshKeyFile,
                                                         final UserVm userVm, final long timeoutTime,
                                                         final int waitDuration) {
-        String hostName = userVm.getHostName();
-        if (StringUtils.isNotEmpty(hostName)) {
-            hostName = hostName.toLowerCase();
-        }
+        String hostName = getKubernetesNodeName(userVm);
         while (System.currentTimeMillis() < timeoutTime) {
             Pair<Boolean, String> result = null;
             try {

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeAccess.java

@@ -0,0 +1,67 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.kubernetes.cluster.utils;
+
+import java.io.File;
+
+/** SSH endpoints for a CKS node: the control plane runs kubectl, the node itself runs guest commands. */
+public final class NodeAccess {
+    private final String controlAddress;
+    private final int controlPort;
+    private final String nodeAddress;
+    private final int nodePort;
+    private final String user;
+    private final File sshKeyFile;
+
+    public NodeAccess(String controlAddress,
+                      int controlPort,
+                      String nodeAddress,
+                      int nodePort,
+                      String user,
+                      File sshKeyFile) {
+        this.controlAddress = controlAddress;
+        this.controlPort = controlPort;
+        this.nodeAddress = nodeAddress;
+        this.nodePort = nodePort;
+        this.user = user;
+        this.sshKeyFile = sshKeyFile;
+    }
+
+    public String getControlAddress() {
+        return controlAddress;
+    }
+
+    public int getControlPort() {
+        return controlPort;
+    }
+
+    public String getNodeAddress() {
+        return nodeAddress;
+    }
+
+    public int getNodePort() {
+        return nodePort;
+    }
+
+    public String getUser() {
+        return user;
+    }
+
+    public File getSshKeyFile() {
+        return sshKeyFile;
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacityReconciliationException.java

@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.kubernetes.cluster.utils;
+
+import com.cloud.utils.exception.CloudRuntimeException;
+
+/** An operational failure while refreshing a node's guest or Kubernetes resources. */
+public class NodeCapacityReconciliationException extends CloudRuntimeException {
+    private static final long serialVersionUID = 1L;
+
+    public NodeCapacityReconciliationException(String message) {
+        super(message);
+    }
+
+    public NodeCapacityReconciliationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeCapacitySnapshot.java

@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.kubernetes.cluster.utils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+
+/** The parts of a Kubernetes Node object that a live resize reads. */
+public final class NodeCapacitySnapshot {
+    /** Set while a node has resize work that has not been verified yet; the value is the cluster UUID. */
+    public static final String PENDING_ANNOTATION = "cloudstack.apache.org/cks-live-resize-pending";
+    /** Set when CKS, not an operator, cordoned the node for a resize; the value is the cluster UUID. */
+    public static final String CORDON_ANNOTATION = "cloudstack.apache.org/cks-live-resize";
+
+    private final String uid;
+    private final String resourceVersion;
+    private final boolean unschedulable;
+    private final Map<String, String> annotations;
+    private final boolean ready;
+    private final NodeResources capacity;
+    private final NodeResources allocatable;
+
+    public NodeCapacitySnapshot(String uid,
+                                String resourceVersion,
+                                boolean unschedulable,
+                                Map<String, String> annotations,
+                                boolean ready,
+                                NodeResources capacity,
+                                NodeResources allocatable) {
+        this.uid = uid;
+        this.resourceVersion = resourceVersion;
+        this.unschedulable = unschedulable;
+        this.annotations = Collections.unmodifiableMap(new HashMap<>(annotations));
+        this.ready = ready;
+        this.capacity = capacity;
+        this.allocatable = allocatable;
+    }
+
+    public String getUid() {
+        return uid;
+    }
+
+    public String getResourceVersion() {
+        return resourceVersion;
+    }
+
+    public boolean isUnschedulable() {
+        return unschedulable;
+    }
+
+    public boolean isReady() {
+        return ready;
+    }
+
+    public NodeResources getCapacity() {
+        return capacity;
+    }
+
+    public NodeResources getAllocatable() {
+        return allocatable;
+    }
+
+    public boolean isPendingFor(String clusterUuid) {
+        return clusterUuid.equals(annotations.get(PENDING_ANNOTATION));
+    }
+
+    /** Whether CKS cordoned the node: it is unschedulable and carries this cluster's ownership marker. */
+    public boolean isCordonOwnedBy(String clusterUuid) {
+        return unschedulable && hasOwnershipMarker(clusterUuid);
+    }
+
+    /**
+     * Whether this cluster's ownership marker is set. kubectl uncordon keeps annotations, so the marker can
+     * be left on a node somebody has uncordoned since; it then no longer means that CKS owns a cordon.
+     */
+    public boolean hasOwnershipMarker(String clusterUuid) {
+        return clusterUuid.equals(annotations.get(CORDON_ANNOTATION));
+    }
+
+    /** Whether a resize marker carries another cluster's UUID; such markers are never changed. */
+    public boolean hasForeignMarker(String clusterUuid) {
+        return isForeign(annotations.get(PENDING_ANNOTATION), clusterUuid)
+                || isForeign(annotations.get(CORDON_ANNOTATION), clusterUuid);
+    }
+
+    private static boolean isForeign(String owner, String clusterUuid) {
+        return StringUtils.isNotEmpty(owner) && !owner.equals(clusterUuid);
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/NodeResources.java

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.kubernetes.cluster.utils;
+
+/** CPU and memory quantities from a Kubernetes Node status, such as its capacity or allocatable. */
+public final class NodeResources {
+    private final long cpuMillis;
+    private final long memoryBytes;
+
+    public NodeResources(long cpuMillis, long memoryBytes) {
+        this.cpuMillis = cpuMillis;
+        this.memoryBytes = memoryBytes;
+    }
+
+    public long getCpuMillis() {
+        return cpuMillis;
+    }
+
+    public long getMemoryBytes() {
+        return memoryBytes;
+    }
+}