andrijapanicsb commented on code in PR #14223:
URL: https://github.com/apache/cloudstack/pull/14223#discussion_r4394973071
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java:
##########
@@ -376,24 +377,171 @@ private void
scaleKubernetesClusterOffering(KubernetesClusterNodeType nodeType,
List<KubernetesClusterVmMapVO> vmList =
kubernetesClusterVmMapDao.listByClusterIdAndVmType(kubernetesCluster.getId(),
nodeType);
final long tobeScaledVMCount = Math.min(vmList.size(), size);
for (long i = 0; i < tobeScaledVMCount; i++) {
- KubernetesClusterVmMapVO vmMapVO = vmList.get((int) i);
- UserVmVO userVM = userVmDao.findById(vmMapVO.getVmId());
- boolean result = false;
- try {
- result = userVmManager.upgradeVirtualMachine(userVM.getId(),
serviceOffering.getId(), new HashMap<String, String>());
- } catch (RuntimeException | ResourceUnavailableException |
ManagementServerException | VirtualMachineMigrationException e) {
- logTransitStateAndThrow(Level.ERROR, String.format("Scaling
Kubernetes cluster : %s failed, unable to scale cluster VM : %s due to %s",
kubernetesCluster.getName(), userVM.getDisplayName(), e.getMessage()),
kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, e);
- }
- if (!result) {
- logTransitStateAndThrow(Level.WARN, String.format("Scaling
Kubernetes cluster : %s failed, unable to scale cluster VM : %s",
kubernetesCluster.getName(),
userVM.getDisplayName()),kubernetesCluster.getId(),
KubernetesCluster.Event.OperationFailed);
- }
+ new NodeOfferingResize(vmList.get((int) i), nodeType,
serviceOffering).execute();
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);
}
}
kubernetesCluster = updateKubernetesClusterEntryForNodeType(null,
nodeType, serviceOffering, updateNodeOffering, updateClusterOffering);
}
+ private final class NodeOfferingResize {
+ private final KubernetesClusterVmMapVO vmMap;
+ private final KubernetesClusterNodeType nodeType;
+ private final ServiceOffering targetOffering;
+ private final ServiceOffering oldOffering;
+ private UserVmVO vm;
+ private NodeAccess access;
+ private NodeCapacitySnapshot before;
+ private boolean reconcileKubernetes;
+ private boolean cksCordonCompleted;
+ private boolean resizeSucceeded;
+
+ private NodeOfferingResize(KubernetesClusterVmMapVO vmMap,
KubernetesClusterNodeType nodeType, ServiceOffering targetOffering) {
+ this.vmMap = vmMap;
+ this.nodeType = nodeType;
+ this.targetOffering = targetOffering;
+ vm = userVmDao.findById(vmMap.getVmId());
+ if (vm == null) {
+ logTransitStateAndThrow(Level.ERROR, String.format("Scaling
Kubernetes cluster : %s failed, unable to find cluster VM %s",
+ kubernetesCluster.getName(), vmMap.getVmId()),
kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed);
+ }
+ oldOffering =
serviceOfferingDao.findById(vm.getServiceOfferingId());
+ resizeSucceeded = Objects.equals(targetOffering.getId(),
vm.getServiceOfferingId());
+ }
+
+ private void execute() {
+ validateLiveResize();
+ try {
+ prepareReconciliation();
+ resizeVm();
+ reconcileCapacity();
+ } catch (ResourceUnavailableException | ManagementServerException
| VirtualMachineMigrationException | CloudRuntimeException e) {
+ restoreAfterFailedResize();
+ reportFailure(e);
+ }
+ }
+
+ private void validateLiveResize() {
+ if
(KubernetesClusterNodeCapacityReconcilerImpl.capacityChanged(oldOffering,
targetOffering)
+ && 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(), vm.getDisplayName()),
+ kubernetesCluster.getId(),
KubernetesCluster.Event.OperationFailed);
+ }
+ }
+
+ private void prepareReconciliation() {
+ reconcileKubernetes = shouldReconcileNodeCapacity(vmMap, vm,
oldOffering, targetOffering, nodeType);
+ if (reconcileKubernetes || isReconciliationRetryCandidate(vmMap,
vm, targetOffering, nodeType)) {
+ access = resolveNodeAccess(vm);
+ before =
kubernetesClusterNodeCapacityReconciler.captureBefore(kubernetesCluster, vm,
access);
+ reconcileKubernetes = reconcileKubernetes ||
before.isCloudStackResizeCordon();
+ }
+ if (reconcileKubernetes) {
+
kubernetesClusterNodeCapacityReconciler.cordonIfNeeded(kubernetesCluster, vm,
before, access, scaleTimeoutTime);
+ cksCordonCompleted = !before.isUnschedulable() ||
before.isCloudStackResizeCordon();
+ }
+ }
+
+ private void resizeVm() throws ResourceUnavailableException,
ManagementServerException, VirtualMachineMigrationException {
+ if (resizeSucceeded) {
+ return;
+ }
+ if (!userVmManager.upgradeVirtualMachine(vm.getId(),
targetOffering.getId(), new HashMap<>())) {
+ throw new CloudRuntimeException("Unable to scale cluster VM "
+ vm.getDisplayName());
+ }
+ resizeSucceeded = true;
+ UserVmVO resizedVm = userVmDao.findById(vm.getId());
+ if (resizedVm == null || !Objects.equals(targetOffering.getId(),
resizedVm.getServiceOfferingId())) {
+ throw new CloudRuntimeException("VM " + vm.getUuid() + " did
not reach target offering");
+ }
+ vm = resizedVm;
+ }
+
+ private void reconcileCapacity() {
+ if (!reconcileKubernetes) {
+ return;
+ }
+
kubernetesClusterNodeCapacityReconciler.onlineGuestResources(access);
+ kubernetesClusterNodeCapacityReconciler.verifyGuestResources(vm,
targetOffering, before, access, scaleTimeoutTime);
+ if
(!kubernetesClusterNodeCapacityReconciler.isKubernetesResourcesCurrent(before,
targetOffering)) {
+ kubernetesClusterNodeCapacityReconciler.restartKubelet(vm,
access, scaleTimeoutTime);
+
kubernetesClusterNodeCapacityReconciler.waitForKubernetesResources(kubernetesCluster,
vm, targetOffering, before, access, scaleTimeoutTime);
+ }
+
kubernetesClusterNodeCapacityReconciler.restoreSchedulability(kubernetesCluster,
vm, before, access, scaleTimeoutTime);
+ }
+
+ private void restoreAfterFailedResize() {
+ if (!reconcileKubernetes || !cksCordonCompleted ||
resizeSucceeded) {
+ return;
+ }
+ try {
+
kubernetesClusterNodeCapacityReconciler.restoreSchedulability(kubernetesCluster,
vm, before, access, scaleTimeoutTime);
+ } catch (CloudRuntimeException cleanupException) {
+ logger.warn("Unable to restore schedulability for CKS node {}
after a pre-resize failure", vm.getUuid(), cleanupException);
+ }
+ }
+
+ private void reportFailure(Exception cause) {
+ 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,
+ vm.getDisplayName(), vm.getUuid(), oldOffering == null ?
null : oldOffering.getId(),
+ targetOffering.getId(), cause.getMessage(), recovery);
+ logTransitStateAndThrow(Level.ERROR, message,
kubernetesCluster.getId(), KubernetesCluster.Event.OperationFailed, cause);
+ }
+ }
+
+ 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()));
+ }
+ 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());
+ }
+
+ 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);
+ }
+
+ 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
Review Comment:
Done, the shared checks are in one method now, which also gives the reason
for the log line. The already-on-target case only adds the check for the
pending annotation.
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterScaleWorker.java:
##########
@@ -605,6 +753,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;
+ }
Review Comment:
Agreed, it belongs in #14138. Since #14138 is still a draft against 4.22, I
kept the guard for now, because without it this PR fails for clusters with no
etcd nodes and no cluster-wide offering. Once the same handling is in main I'll
check it and remove the guard here, or rebase onto #14138 if it lands first.
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java:
##########
@@ -0,0 +1,68 @@
+// 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 {
Review Comment:
Fair point. The interface is gone, it's one concrete class now, and the
access and snapshot classes are top-level.
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/utils/KubernetesClusterNodeCapacityReconciler.java:
##########
@@ -0,0 +1,68 @@
+// 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);
+ boolean requiresKubeletRefresh(ServiceOffering oldOffering,
ServiceOffering newOffering,
+ KubernetesClusterNodeType nodeType, KubernetesCluster.State
originalState);
+ void cordonIfNeeded(KubernetesCluster cluster, UserVm vm,
NodeCapacitySnapshot before, NodeAccess access, long deadline);
+ void onlineGuestResources(NodeAccess access);
+ void verifyGuestResources(UserVm vm, ServiceOffering target,
NodeCapacitySnapshot before, NodeAccess access, long deadline);
+ boolean isKubernetesResourcesCurrent(NodeCapacitySnapshot snapshot,
ServiceOffering target);
+ void restartKubelet(UserVm vm, NodeAccess access, long deadline);
+ NodeCapacitySnapshot waitForKubernetesResources(KubernetesCluster cluster,
UserVm vm, ServiceOffering target,
+ NodeCapacitySnapshot before, NodeAccess access, long deadline);
+ void restoreSchedulability(KubernetesCluster cluster, UserVm vm,
NodeCapacitySnapshot before, NodeAccess access, long deadline);
+
+ final class NodeAccess {
Review Comment:
Done for both classes.
--
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]
