(WIP) CKS: Allow affinity group selection during cluster creation

api/src/main/java/com/cloud/kubernetes/cluster/KubernetesServiceHelper.java

@@ -18,6 +18,7 @@
 
 import org.apache.cloudstack.acl.ControlledEntity;
 
+import java.util.List;
 import java.util.Map;
 
 import com.cloud.user.Account;
@@ -36,5 +37,6 @@ enum KubernetesClusterNodeType {
     boolean isValidNodeType(String nodeType);
     Map<String, Long> getServiceOfferingNodeTypeMap(Map<String, Map<String, String>> serviceOfferingNodeTypeMap);
     Map<String, Long> getTemplateNodeTypeMap(Map<String, Map<String, String>> templateNodeTypeMap);
+    Map<String, List<Long>> getAffinityGroupNodeTypeMap(Map<String, Map<String, String>> affinityGroupNodeTypeMap);
     void cleanupForAccount(Account account);
 }

api/src/main/java/com/cloud/vm/VmDetailConstants.java

@@ -93,6 +93,7 @@ public interface VmDetailConstants {
     String CKS_NODE_TYPE = "node";
     String OFFERING = "offering";
     String TEMPLATE = "template";
+    String AFFINITY_GROUP = "affinitygroup";
 
     // VMware to KVM VM migrations specific
     String VMWARE_TO_KVM_PREFIX = "vmware-to-kvm";

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -1236,6 +1236,13 @@ public class ApiConstants {
     public static final String MAX_SIZE = "maxsize";
     public static final String NODE_TYPE_OFFERING_MAP = "nodeofferings";
     public static final String NODE_TYPE_TEMPLATE_MAP = "nodetemplates";
+    public static final String NODE_TYPE_AFFINITY_GROUP_MAP = "nodeaffinitygroups";
+    public static final String CONTROL_AFFINITY_GROUP_IDS = "controlaffinitygroupids";
+    public static final String CONTROL_AFFINITY_GROUP_NAMES = "controlaffinitygroupnames";
+    public static final String WORKER_AFFINITY_GROUP_IDS = "workeraffinitygroupids";
+    public static final String WORKER_AFFINITY_GROUP_NAMES = "workeraffinitygroupnames";
+    public static final String ETCD_AFFINITY_GROUP_IDS = "etcdaffinitygroupids";
+    public static final String ETCD_AFFINITY_GROUP_NAMES = "etcdaffinitygroupnames";
 
     public static final String BOOT_TYPE = "boottype";
     public static final String BOOT_MODE = "bootmode";

engine/schema/src/main/resources/META-INF/db/schema-42210to42300.sql

@@ -24,6 +24,19 @@
 UPDATE `cloud`.`configuration` SET value='random' WHERE name IN ('vm.allocation.algorithm', 'volume.allocation.algorithm') AND value='userconcentratedpod_random';
 UPDATE `cloud`.`configuration` SET value='firstfit' WHERE name IN ('vm.allocation.algorithm', 'volume.allocation.algorithm') AND value='userconcentratedpod_firstfit';
 
+-- Create kubernetes_cluster_affinity_group_map table for CKS per-node-type affinity groups
+CREATE TABLE IF NOT EXISTS `cloud`.`kubernetes_cluster_affinity_group_map` (
+    `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+    `cluster_id` bigint unsigned NOT NULL COMMENT 'kubernetes cluster id',
+    `node_type` varchar(32) NOT NULL COMMENT 'CONTROL, WORKER, or ETCD',
+    `affinity_group_id` bigint unsigned NOT NULL COMMENT 'affinity group id',
+    PRIMARY KEY (`id`),
+    CONSTRAINT `fk_kubernetes_cluster_ag_map__cluster_id` FOREIGN KEY (`cluster_id`) REFERENCES `kubernetes_cluster`(`id`) ON DELETE CASCADE,
+    CONSTRAINT `fk_kubernetes_cluster_ag_map__ag_id` FOREIGN KEY (`affinity_group_id`) REFERENCES `affinity_group`(`id`) ON DELETE CASCADE,
+    INDEX `i_kubernetes_cluster_ag_map__cluster_id`(`cluster_id`),
+    INDEX `i_kubernetes_cluster_ag_map__ag_id`(`affinity_group_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
 -- Create webhook_filter table
 DROP TABLE IF EXISTS `cloud`.`webhook_filter`;
 CREATE TABLE IF NOT EXISTS `cloud`.`webhook_filter` (

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

@@ -0,0 +1,83 @@
+// 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;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+import org.apache.cloudstack.api.InternalIdentity;
+
+@Entity
+@Table(name = "kubernetes_cluster_affinity_group_map")
+public class KubernetesClusterAffinityGroupMapVO implements InternalIdentity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "id")
+    private Long id;
+
+    @Column(name = "cluster_id")
+    private long clusterId;
+
+    @Column(name = "node_type")
+    private String nodeType;
+
+    @Column(name = "affinity_group_id")
+    private long affinityGroupId;
+
+    public KubernetesClusterAffinityGroupMapVO() {
+    }
+
+    public KubernetesClusterAffinityGroupMapVO(long clusterId, String nodeType, long affinityGroupId) {
+        this.clusterId = clusterId;
+        this.nodeType = nodeType;
+        this.affinityGroupId = affinityGroupId;
+    }
+
+    @Override
+    public long getId() {
+        return id;
+    }
+
+    public long getClusterId() {
+        return clusterId;
+    }
+
+    public void setClusterId(long clusterId) {
+        this.clusterId = clusterId;
+    }
+
+    public String getNodeType() {
+        return nodeType;
+    }
+
+    public void setNodeType(String nodeType) {
+        this.nodeType = nodeType;
+    }
+
+    public long getAffinityGroupId() {
+        return affinityGroupId;
+    }
+
+    public void setAffinityGroupId(long affinityGroupId) {
+        this.affinityGroupId = affinityGroupId;
+    }
+}

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

@@ -169,6 +169,7 @@
 import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterStartWorker;
 import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterStopWorker;
 import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterUpgradeWorker;
+import com.cloud.kubernetes.cluster.dao.KubernetesClusterAffinityGroupMapDao;
 import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao;
 import com.cloud.kubernetes.cluster.dao.KubernetesClusterDetailsDao;
 import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao;
@@ -315,6 +316,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     @Inject
     public KubernetesClusterDetailsDao kubernetesClusterDetailsDao;
     @Inject
+    public KubernetesClusterAffinityGroupMapDao kubernetesClusterAffinityGroupMapDao;
+    @Inject
     public KubernetesSupportedVersionDao kubernetesSupportedVersionDao;
     @Inject
     protected SSHKeyPairDao sshKeyPairDao;
@@ -858,24 +861,38 @@ public KubernetesClusterResponse createKubernetesClusterResponse(long kubernetes
 
         List<KubernetesUserVmResponse> vmResponses = new ArrayList<>();
         List<KubernetesClusterVmMapVO> vmList = kubernetesClusterVmMapDao.listByClusterId(kubernetesCluster.getId());
-        ResponseView respView = ResponseView.Restricted;
+        ResponseView userVmResponseView = ResponseView.Restricted;
         Account caller = CallContext.current().getCallingAccount();
         if (accountService.isRootAdmin(caller.getId())) {
-            respView = ResponseView.Full;
+            userVmResponseView = ResponseView.Full;
         }
         final String responseName = "virtualmachine";
         if (vmList != null && !vmList.isEmpty()) {
-            for (KubernetesClusterVmMapVO vmMapVO : vmList) {
-                UserVmJoinVO userVM = userVmJoinDao.findById(vmMapVO.getVmId());
-                if (userVM != null) {
-                    UserVmResponse vmResponse = ApiDBUtils.newUserVmResponse(respView, responseName, userVM,
-                        EnumSet.of(VMDetails.nics), caller);
+            Map<Long, KubernetesClusterVmMapVO> vmMapById = vmList.stream()
+                    .collect(Collectors.toMap(KubernetesClusterVmMapVO::getVmId, vm -> vm));
+            Long[] vmIds = vmMapById.keySet().toArray(new Long[0]);
+            List<UserVmJoinVO> userVmJoinVOs = userVmJoinDao.searchByIds(vmIds);
+            if (userVmJoinVOs != null && !userVmJoinVOs.isEmpty()) {
+                Map<Long, UserVmResponse> vmResponseMap = new HashMap<>();
+                for (UserVmJoinVO userVM : userVmJoinVOs) {
+                    Long vmId = userVM.getId();
+                    UserVmResponse vmResponse = vmResponseMap.get(vmId);
+                    if (vmResponse == null) {
+                        vmResponse = ApiDBUtils.newUserVmResponse(userVmResponseView, responseName, userVM,
+                                EnumSet.of(VMDetails.nics, VMDetails.affgrp), caller);
+                        vmResponseMap.put(vmId, vmResponse);
+                    } else {
+                        ApiDBUtils.fillVmDetails(userVmResponseView, vmResponse, userVM);
+                    }
+                }
+                for (Map.Entry<Long, UserVmResponse> vmIdResponseEntry : vmResponseMap.entrySet()) {
                     KubernetesUserVmResponse kubernetesUserVmResponse = new KubernetesUserVmResponse();
                     try {
-                        BeanUtils.copyProperties(kubernetesUserVmResponse, vmResponse);
+                        BeanUtils.copyProperties(kubernetesUserVmResponse, vmIdResponseEntry.getValue());
                     } catch (IllegalAccessException | InvocationTargetException e) {
                         throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Failed to generate zone metrics response");
                     }
+                    KubernetesClusterVmMapVO vmMapVO = vmMapById.get(vmIdResponseEntry.getKey());
                     kubernetesUserVmResponse.setExternalNode(vmMapVO.isExternalNode());
                     kubernetesUserVmResponse.setEtcdNode(vmMapVO.isEtcdNode());
                     kubernetesUserVmResponse.setNodeVersion(vmMapVO.getNodeVersion());
@@ -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()) {
+            return;
+        }
+        List<String> affinityGroupUuids = new ArrayList<>();
+        List<String> affinityGroupNames = new ArrayList<>();
+        for (Long affinityGroupId : affinityGroupIds) {
+            AffinityGroupVO affinityGroup = affinityGroupDao.findById(affinityGroupId);
+            if (affinityGroup != null) {
+                affinityGroupUuids.add(affinityGroup.getUuid());
+                affinityGroupNames.add(affinityGroup.getName());
+            }
+        }
+        String affinityGroupUuidsCsv = String.join(",", affinityGroupUuids);
+        String affinityGroupNamesCsv = String.join(",", affinityGroupNames);
+        if (CONTROL.name().equals(nodeType)) {
+            response.setControlAffinityGroupIds(affinityGroupUuidsCsv);
+            response.setControlAffinityGroupNames(affinityGroupNamesCsv);
+        } else if (WORKER.name().equals(nodeType)) {
+            response.setWorkerAffinityGroupIds(affinityGroupUuidsCsv);
+            response.setWorkerAffinityGroupNames(affinityGroupNamesCsv);
+        } else if (ETCD.name().equals(nodeType)) {
+            response.setEtcdAffinityGroupIds(affinityGroupUuidsCsv);
+            response.setEtcdAffinityGroupNames(affinityGroupNamesCsv);
+        }
+    }
+
     private DataCenter validateAndGetZoneForKubernetesCreateParameters(Long zoneId, Long networkId) {
         DataCenter zone = dataCenterDao.findById(zoneId);
         if (zone == null) {
@@ -1187,6 +1239,20 @@ private Network getKubernetesClusterNetworkIfMissing(final String clusterName, f
         return network;
     }
 
+    private void persistAffinityGroupMappings(long clusterId, Map<String, List<Long>> affinityGroupNodeTypeMap) {
+        if (MapUtils.isEmpty(affinityGroupNodeTypeMap)) {
+            return;
+        }
+        for (Map.Entry<String, List<Long>> nodeTypeAffinityGroupEntry : affinityGroupNodeTypeMap.entrySet()) {
+            String nodeType = nodeTypeAffinityGroupEntry.getKey();
+            List<Long> affinityGroupIds = nodeTypeAffinityGroupEntry.getValue();
+            for (Long affinityGroupId : affinityGroupIds) {
+                kubernetesClusterAffinityGroupMapDao.persist(
+                    new KubernetesClusterAffinityGroupMapVO(clusterId, nodeType, affinityGroupId));
+            }
+        }
+    }
+
     private void addKubernetesClusterDetails(final KubernetesCluster kubernetesCluster, final Network network, final CreateKubernetesClusterCmd cmd) {
         final String externalLoadBalancerIpAddress = cmd.getExternalLoadBalancerIpAddress();
         final String dockerRegistryUserName = cmd.getDockerRegistryUserName();
@@ -1627,6 +1693,7 @@ public KubernetesCluster createManagedKubernetesCluster(CreateKubernetesClusterC
         }
 
         Map<String, Long> templateNodeTypeMap = cmd.getTemplateNodeTypeMap();
+        Map<String, List<Long>> affinityGroupNodeTypeMap = cmd.getAffinityGroupNodeTypeMap();
         final VMTemplateVO finalTemplate = getKubernetesServiceTemplate(zone, hypervisorType, templateNodeTypeMap, DEFAULT, clusterKubernetesVersion);
         final VMTemplateVO controlNodeTemplate = getKubernetesServiceTemplate(zone, hypervisorType, templateNodeTypeMap, CONTROL, clusterKubernetesVersion);
         final VMTemplateVO workerNodeTemplate = getKubernetesServiceTemplate(zone, hypervisorType, templateNodeTypeMap, WORKER, clusterKubernetesVersion);
@@ -1672,6 +1739,7 @@ public KubernetesClusterVO doInTransaction(TransactionStatus status) {
                 }
                 newCluster.setCsiEnabled(cmd.getEnableCsi());
                 kubernetesClusterDao.persist(newCluster);
+                persistAffinityGroupMappings(newCluster.getId(), affinityGroupNodeTypeMap);
                 addKubernetesClusterDetails(newCluster, defaultNetwork, cmd);
                 return newCluster;
             }

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

@@ -18,12 +18,16 @@
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 import javax.inject.Inject;
 
+import org.apache.cloudstack.affinity.AffinityGroup;
+import org.apache.cloudstack.affinity.dao.AffinityGroupDao;
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.offering.ServiceOffering;
 import com.cloud.service.dao.ServiceOfferingDao;
@@ -66,6 +70,8 @@ public class KubernetesServiceHelperImpl extends AdapterBase implements Kubernet
     @Inject
     protected VMTemplateDao vmTemplateDao;
     @Inject
+    protected AffinityGroupDao affinityGroupDao;
+    @Inject
     KubernetesClusterService kubernetesClusterService;
 
     protected void setEventTypeEntityDetails(Class<?> eventTypeDefinedClass, Class<?> entityClass) {
@@ -244,6 +250,81 @@ public Map<String, Long> getTemplateNodeTypeMap(Map<String, Map<String, String>>
         return mapping;
     }
 
+    protected void checkNodeTypeAffinityGroupEntryCompleteness(String nodeType, String affinityGroupUuids) {
+        if (StringUtils.isAnyBlank(nodeType, affinityGroupUuids)) {
+            String error = String.format("Any Node Type to Affinity Group entry should have a valid '%s' and '%s' values",
+                    VmDetailConstants.CKS_NODE_TYPE, VmDetailConstants.AFFINITY_GROUP);
+            logger.error(error);
+            throw new InvalidParameterValueException(error);
+        }
+    }
+
+    protected void checkNodeTypeAffinityGroupEntryNodeType(String nodeType) {
+        if (!isValidNodeType(nodeType)) {
+            String error = String.format("The provided value '%s' for Node Type is invalid", nodeType);
+            logger.error(error);
+            throw new InvalidParameterValueException(error);
+        }
+    }
+
+    protected Long validateAffinityGroupUuidAndGetId(String affinityGroupUuid) {
+        if (StringUtils.isBlank(affinityGroupUuid)) {
+            String error = "Empty affinity group UUID provided";
+            logger.error(error);
+            throw new InvalidParameterValueException(error);
+        }
+        AffinityGroup affinityGroup = affinityGroupDao.findByUuid(affinityGroupUuid);
+        if (affinityGroup == null) {
+            String error = String.format("Cannot find an affinity group with ID %s", affinityGroupUuid);
+            logger.error(error);
+            throw new InvalidParameterValueException(error);
+        }
+        return affinityGroup.getId();
+    }
+
+    protected List<Long> validateAndGetAffinityGroupIds(String affinityGroupUuids) {
+        String[] uuids = affinityGroupUuids.split(",");
+        List<Long> affinityGroupIds = new ArrayList<>();
+        for (String uuid : uuids) {
+            String trimmedUuid = uuid.trim();
+            Long affinityGroupId = validateAffinityGroupUuidAndGetId(trimmedUuid);
+            affinityGroupIds.add(affinityGroupId);
+        }
+        return affinityGroupIds;
+    }
+
+    protected void addNodeTypeAffinityGroupEntry(String nodeType, List<Long> affinityGroupIds, Map<String, List<Long>> nodeTypeToAffinityGroupIds) {
+        if (logger.isDebugEnabled()) {
+            logger.debug(String.format("Node Type: '%s' should use affinity group IDs: '%s'", nodeType, affinityGroupIds));
+        }
+        KubernetesClusterNodeType clusterNodeType = KubernetesClusterNodeType.valueOf(nodeType.toUpperCase());
+        nodeTypeToAffinityGroupIds.put(clusterNodeType.name(), affinityGroupIds);
+    }
+
+    protected void processNodeTypeAffinityGroupEntryAndAddToMappingIfValid(Map<String, String> nodeTypeAffinityConfig, Map<String, List<Long>> nodeTypeToAffinityGroupIds) {
+        if (MapUtils.isEmpty(nodeTypeAffinityConfig)) {
+            return;
+        }
+        String nodeType = nodeTypeAffinityConfig.get(VmDetailConstants.CKS_NODE_TYPE);
+        String affinityGroupUuids = nodeTypeAffinityConfig.get(VmDetailConstants.AFFINITY_GROUP);
+        checkNodeTypeAffinityGroupEntryCompleteness(nodeType, affinityGroupUuids);
+        checkNodeTypeAffinityGroupEntryNodeType(nodeType);
+
+        List<Long> affinityGroupIds = validateAndGetAffinityGroupIds(affinityGroupUuids);
+        addNodeTypeAffinityGroupEntry(nodeType, affinityGroupIds, nodeTypeToAffinityGroupIds);
+    }
+
+    @Override
+    public Map<String, List<Long>> getAffinityGroupNodeTypeMap(Map<String, Map<String, String>> affinityGroupNodeTypeMap) {
+        Map<String, List<Long>> nodeTypeToAffinityGroupIds = new HashMap<>();
+        if (MapUtils.isNotEmpty(affinityGroupNodeTypeMap)) {
+            for (Map<String, String> nodeTypeAffinityConfig : affinityGroupNodeTypeMap.values()) {
+                processNodeTypeAffinityGroupEntryAndAddToMappingIfValid(nodeTypeAffinityConfig, nodeTypeToAffinityGroupIds);
+            }
+        }
+        return nodeTypeToAffinityGroupIds;
+    }
+
     public void cleanupForAccount(Account account) {
         kubernetesClusterService.cleanupForAccount(account);
     }