nvazquez commented on code in PR #12386:
URL: https://github.com/apache/cloudstack/pull/12386#discussion_r2802056027
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterActionWorker.java:
##########
@@ -1112,4 +1115,18 @@ public Long getExplicitAffinityGroup(Long domainId, Long
accountId) {
}
return null;
}
+
+ protected List<Long>
getAffinityGroupIdsForNodeType(KubernetesClusterNodeType nodeType) {
+ return new
ArrayList<>(kubernetesClusterAffinityGroupMapDao.listAffinityGroupIdsByClusterIdAndNodeType(
Review Comment:
Here as well, can we add checks for null or empty list?
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java:
##########
@@ -905,10 +922,45 @@ public KubernetesClusterResponse
createKubernetesClusterResponse(long kubernetes
response.setClusterType(kubernetesCluster.getClusterType());
response.setCsiEnabled(kubernetesCluster.isCsiEnabled());
response.setCreated(kubernetesCluster.getCreated());
+ setNodeTypeAffinityGroupResponse(response, kubernetesCluster.getId());
return response;
}
+ protected void setNodeTypeAffinityGroupResponse(KubernetesClusterResponse
response, long clusterId) {
+ setAffinityGroupResponseForNodeType(response, clusterId,
CONTROL.name());
+ setAffinityGroupResponseForNodeType(response, clusterId,
WORKER.name());
+ setAffinityGroupResponseForNodeType(response, clusterId, ETCD.name());
+ }
+
+ protected void
setAffinityGroupResponseForNodeType(KubernetesClusterResponse response, long
clusterId, String nodeType) {
+ List<Long> affinityGroupIds =
kubernetesClusterAffinityGroupMapDao.listAffinityGroupIdsByClusterIdAndNodeType(clusterId,
nodeType);
+ if (affinityGroupIds == null || affinityGroupIds.isEmpty()) {
Review Comment:
Minor: can use `CollectionUtils.isEmpty(affinityGroupIds)` here
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterAffinityGroupMapDaoImpl.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.dao;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.springframework.stereotype.Component;
+
+import com.cloud.kubernetes.cluster.KubernetesClusterAffinityGroupMapVO;
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+
+@Component
+public class KubernetesClusterAffinityGroupMapDaoImpl extends
GenericDaoBase<KubernetesClusterAffinityGroupMapVO, Long>
+ implements KubernetesClusterAffinityGroupMapDao {
+
+ private final SearchBuilder<KubernetesClusterAffinityGroupMapVO>
clusterIdAndNodeTypeSearch;
+ private final SearchBuilder<KubernetesClusterAffinityGroupMapVO>
clusterIdSearch;
+
+ public KubernetesClusterAffinityGroupMapDaoImpl() {
+ clusterIdAndNodeTypeSearch = createSearchBuilder();
+ clusterIdAndNodeTypeSearch.and("clusterId",
clusterIdAndNodeTypeSearch.entity().getClusterId(), SearchCriteria.Op.EQ);
+ clusterIdAndNodeTypeSearch.and("nodeType",
clusterIdAndNodeTypeSearch.entity().getNodeType(), SearchCriteria.Op.EQ);
+ clusterIdAndNodeTypeSearch.done();
+
+ clusterIdSearch = createSearchBuilder();
+ clusterIdSearch.and("clusterId",
clusterIdSearch.entity().getClusterId(), SearchCriteria.Op.EQ);
+ clusterIdSearch.done();
+ }
+
+ @Override
+ public List<KubernetesClusterAffinityGroupMapVO>
listByClusterIdAndNodeType(long clusterId, String nodeType) {
+ SearchCriteria<KubernetesClusterAffinityGroupMapVO> sc =
clusterIdAndNodeTypeSearch.create();
+ sc.setParameters("clusterId", clusterId);
+ sc.setParameters("nodeType", nodeType);
+ return listBy(sc);
+ }
+
+ @Override
+ public List<Long> listAffinityGroupIdsByClusterIdAndNodeType(long
clusterId, String nodeType) {
+ List<KubernetesClusterAffinityGroupMapVO> maps =
listByClusterIdAndNodeType(clusterId, nodeType);
+ return
maps.stream().map(KubernetesClusterAffinityGroupMapVO::getAffinityGroupId).collect(Collectors.toList());
Review Comment:
I think a null check should be needed at this point, to cover cases in which
the affinity group is not passed for the node type
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java:
##########
@@ -2348,6 +2417,100 @@ private List<Long> validateNodes(List<Long> nodeIds,
Long networkId, String netw
return validNodeIds;
}
+ protected void validateNodeAffinityGroups(List<Long> nodeIds,
KubernetesCluster cluster) {
+ List<Long> workerAffinityGroupIds =
kubernetesClusterAffinityGroupMapDao.listAffinityGroupIdsByClusterIdAndNodeType(
+ cluster.getId(), WORKER.name());
+ if (CollectionUtils.isEmpty(workerAffinityGroupIds)) {
+ return;
+ }
+
+ Set<Long> existingWorkerHostIds = getExistingWorkerHostIds(cluster);
+
+ for (Long affinityGroupId : workerAffinityGroupIds) {
+ AffinityGroupVO affinityGroup =
affinityGroupDao.findById(affinityGroupId);
+ if (affinityGroup == null) {
+ continue;
+ }
+
+ validateNodesAgainstExistingWorkers(nodeIds,
existingWorkerHostIds, affinityGroup, cluster);
+ validateNewNodesAntiAffinity(nodeIds, affinityGroup, cluster);
+ }
+ }
+
+ protected Set<Long> getExistingWorkerHostIds(KubernetesCluster cluster) {
+ List<KubernetesClusterVmMapVO> existingWorkerVms =
kubernetesClusterVmMapDao.listByClusterIdAndVmType(cluster.getId(), WORKER);
+ Set<Long> existingWorkerHostIds = new HashSet<>();
+ for (KubernetesClusterVmMapVO workerVmMap : existingWorkerVms) {
+ VMInstanceVO workerVm =
vmInstanceDao.findById(workerVmMap.getVmId());
+ if (workerVm != null && workerVm.getHostId() != null) {
+ existingWorkerHostIds.add(workerVm.getHostId());
+ }
+ }
+ return existingWorkerHostIds;
+ }
+
+ protected void validateNodesAgainstExistingWorkers(List<Long> nodeIds,
Set<Long> existingWorkerHostIds,
+ AffinityGroupVO
affinityGroup, KubernetesCluster cluster) {
+ String affinityGroupType = affinityGroup.getType();
+
+ for (Long nodeId : nodeIds) {
+ VMInstanceVO node = vmInstanceDao.findById(nodeId);
+ if (node == null || node.getHostId() == null) {
+ continue;
+ }
+ Long nodeHostId = node.getHostId();
+ String nodeHostName = getHostName(nodeHostId);
+
+ if ("host anti-affinity".equalsIgnoreCase(affinityGroupType)) {
Review Comment:
What about adding the following checks to not rely on a string comparisson?
- Inject `AffinityGroupService` on the class level
- Add the following method on `AffinityGroupService` interface: `protected
Map<String, AffinityGroupProcessor> getAffinityTypeToProcessorMap()`
- Mark this method as public on `AffinityGroupServiceImpl` (definition
already exists)
- Obtain the `AffinityGroupProcessor` for the affinity group type (adding
null checks) and replace the string comparisson to `processor instanceof
HostAntiAffinityProcessor` and `processor isinstanceof HostAffinityProcessor`
What do you think?
##########
plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java:
##########
@@ -2348,6 +2417,100 @@ private List<Long> validateNodes(List<Long> nodeIds,
Long networkId, String netw
return validNodeIds;
}
+ protected void validateNodeAffinityGroups(List<Long> nodeIds,
KubernetesCluster cluster) {
+ List<Long> workerAffinityGroupIds =
kubernetesClusterAffinityGroupMapDao.listAffinityGroupIdsByClusterIdAndNodeType(
+ cluster.getId(), WORKER.name());
+ if (CollectionUtils.isEmpty(workerAffinityGroupIds)) {
+ return;
+ }
+
+ Set<Long> existingWorkerHostIds = getExistingWorkerHostIds(cluster);
+
+ for (Long affinityGroupId : workerAffinityGroupIds) {
+ AffinityGroupVO affinityGroup =
affinityGroupDao.findById(affinityGroupId);
+ if (affinityGroup == null) {
+ continue;
+ }
+
+ validateNodesAgainstExistingWorkers(nodeIds,
existingWorkerHostIds, affinityGroup, cluster);
+ validateNewNodesAntiAffinity(nodeIds, affinityGroup, cluster);
+ }
+ }
+
+ protected Set<Long> getExistingWorkerHostIds(KubernetesCluster cluster) {
+ List<KubernetesClusterVmMapVO> existingWorkerVms =
kubernetesClusterVmMapDao.listByClusterIdAndVmType(cluster.getId(), WORKER);
+ Set<Long> existingWorkerHostIds = new HashSet<>();
+ for (KubernetesClusterVmMapVO workerVmMap : existingWorkerVms) {
+ VMInstanceVO workerVm =
vmInstanceDao.findById(workerVmMap.getVmId());
+ if (workerVm != null && workerVm.getHostId() != null) {
+ existingWorkerHostIds.add(workerVm.getHostId());
+ }
+ }
+ return existingWorkerHostIds;
+ }
+
+ protected void validateNodesAgainstExistingWorkers(List<Long> nodeIds,
Set<Long> existingWorkerHostIds,
+ AffinityGroupVO
affinityGroup, KubernetesCluster cluster) {
+ String affinityGroupType = affinityGroup.getType();
+
+ for (Long nodeId : nodeIds) {
+ VMInstanceVO node = vmInstanceDao.findById(nodeId);
+ if (node == null || node.getHostId() == null) {
+ continue;
+ }
+ Long nodeHostId = node.getHostId();
+ String nodeHostName = getHostName(nodeHostId);
+
+ if ("host anti-affinity".equalsIgnoreCase(affinityGroupType)) {
+ if (existingWorkerHostIds.contains(nodeHostId)) {
+ throw new InvalidParameterValueException(String.format(
+ "Cannot add VM %s to cluster %s. VM is running on
host %s which violates the cluster's " +
+ "host anti-affinity rule (affinity group: %s).
Existing worker VMs are already running on this host.",
+ node.getInstanceName(), cluster.getName(),
nodeHostName, affinityGroup.getName()));
+ }
+ } else if ("host affinity".equalsIgnoreCase(affinityGroupType)) {
+ if (!existingWorkerHostIds.isEmpty() &&
!existingWorkerHostIds.contains(nodeHostId)) {
+ List<String> existingHostNames =
existingWorkerHostIds.stream()
+ .map(this::getHostName)
+ .collect(Collectors.toList());
+ throw new InvalidParameterValueException(String.format(
+ "Cannot add VM %s to cluster %s. VM is running on
host %s which violates the cluster's " +
+ "host affinity rule (affinity group: %s). All
worker VMs must run on the same host. " +
+ "Existing workers are on host(s): %s.",
+ node.getInstanceName(), cluster.getName(),
nodeHostName, affinityGroup.getName(),
+ String.join(", ", existingHostNames)));
+ }
+ }
+ }
+ }
+
+ protected void validateNewNodesAntiAffinity(List<Long> nodeIds,
AffinityGroupVO affinityGroup, KubernetesCluster cluster) {
+ if (!"host anti-affinity".equalsIgnoreCase(affinityGroup.getType())) {
Review Comment:
Same here
--
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]
