implement node affinity group validation method

Author: Daman Arora

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

@@ -2357,6 +2357,7 @@ public boolean addNodesToKubernetesCluster(AddNodesToKubernetesClusterCmd cmd) {
         if (validNodeIds.isEmpty()) {
             throw new CloudRuntimeException("No valid nodes found to be added to the Kubernetes cluster");
         }
+        validateNodeAffinityGroups(validNodeIds, kubernetesCluster);
         KubernetesClusterAddWorker addWorker = new KubernetesClusterAddWorker(kubernetesCluster, KubernetesClusterManagerImpl.this);
         addWorker = ComponentContext.inject(addWorker);
         return addWorker.addNodesToCluster(validNodeIds, cmd.isMountCksIsoOnVr(), cmd.isManualUpgrade());
@@ -2416,6 +2417,83 @@ 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;
+        }
+
+        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());
+            }
+        }
+
+        for (Long affinityGroupId : workerAffinityGroupIds) {
+            AffinityGroupVO affinityGroup = affinityGroupDao.findById(affinityGroupId);
+            if (affinityGroup == null) {
+                continue;
+            }
+            String affinityGroupType = affinityGroup.getType();
+
+            for (Long nodeId : nodeIds) {
+                VMInstanceVO node = vmInstanceDao.findById(nodeId);
+                if (node == null || node.getHostId() == null) {
+                    continue;
+                }
+                Long nodeHostId = node.getHostId();
+                HostVO nodeHost = hostDao.findById(nodeHostId);
+                String nodeHostName = nodeHost != null ? nodeHost.getName() : String.valueOf(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 = new ArrayList<>();
+                        for (Long hostId : existingWorkerHostIds) {
+                            HostVO host = hostDao.findById(hostId);
+                            existingHostNames.add(host != null ? host.getName() : String.valueOf(hostId));
+                        }
+                        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)));
+                    }
+                }
+            }
+
+            if ("host anti-affinity".equalsIgnoreCase(affinityGroupType)) {
+                Set<Long> newNodeHostIds = new HashSet<>();
+                for (Long nodeId : nodeIds) {
+                    VMInstanceVO node = vmInstanceDao.findById(nodeId);
+                    if (node != null && node.getHostId() != null) {
+                        Long nodeHostId = node.getHostId();
+                        if (newNodeHostIds.contains(nodeHostId)) {
+                            HostVO nodeHost = hostDao.findById(nodeHostId);
+                            String nodeHostName = nodeHost != null ? nodeHost.getName() : String.valueOf(nodeHostId);
+                            throw new InvalidParameterValueException(String.format(
+                                    "Cannot add VM %s to cluster %s. Multiple VMs being added are running on the same host %s, " +
+                                    "which violates the cluster's host anti-affinity rule (affinity group: %s).",
+                                    node.getInstanceName(), cluster.getName(), nodeHostName, affinityGroup.getName()));
+                        }
+                        newNodeHostIds.add(nodeHostId);
+                    }
+                }
+            }
+        }
+    }
+
     @Override
     public List<RemoveVirtualMachinesFromKubernetesClusterResponse> removeVmsFromCluster(RemoveVirtualMachinesFromKubernetesClusterCmd cmd) {
         if (!KubernetesServiceEnabled.value()) {