Re-check affinity before executing each DRS migration

A DRS plan is generated once and executed later, and nothing downstream
re-checks it - migrateVirtualMachine does not enforce affinity groups.
By execution time the cluster may have changed, so a plan that was valid
when generated can violate anti-affinity when it runs.

- validate each migration against current state before queueing it
- skip and mark failed instead of migrating into a violation
- track destinations already queued in this run, since the jobs are
  asynchronous and the database does not reflect them yet

Signed-off-by: Brad House <bhouse@nexthop.ai>

Author: bhouse-nexthop

server/src/main/java/org/apache/cloudstack/cluster/ClusterDrsServiceImpl.java

@@ -768,6 +768,29 @@ void processPlans() {
      * @param plan
      *         the DRS plan to be executed
      */
+    /**
+     * Checks a planned migration against the affinity rules as they stand now.
+     *
+     * A plan is generated once and executed later, so state can have moved on: VMs may have been
+     * created, migrated or destroyed in between. Anti-affinity in particular is only meaningful
+     * against current placements, and nothing downstream re-checks it - migrateVirtualMachine does
+     * not enforce affinity groups.
+     *
+     * @param dispatched
+     *         migrations already queued by this run, which the database does not reflect yet
+     */
+    protected boolean destinationViolatesAffinity(VirtualMachine vm, Host destHost, List<VirtualMachine> dispatched) {
+        if (CollectionUtils.isEmpty(affinityGroupVMMapDao.listByInstanceId(vm.getId()))) {
+            return false;
+        }
+        DataCenterDeployment plan = new DataCenterDeployment(destHost.getDataCenterId(), destHost.getPodId(),
+                destHost.getClusterId(), null, null, null);
+        VirtualMachineProfile vmProfile = new VirtualMachineProfileImpl(vm, null,
+                serviceOfferingDao.findByIdIncludingRemoved(vm.getId(), vm.getServiceOfferingId()), null, null);
+        ExcludeList excludes = managementServer.applyAffinityConstraints(vm, vmProfile, plan, dispatched);
+        return excludes.shouldAvoid(destHost);
+    }
+
     void executeDrsPlan(ClusterDrsPlanVO plan) {
         List<ClusterDrsPlanMigrationVO> planMigrations = drsPlanMigrationDao.listPlanMigrationsToExecute(plan.getId());
         if (planMigrations == null || planMigrations.isEmpty()) {
@@ -783,21 +806,36 @@ void executeDrsPlan(ClusterDrsPlanVO plan) {
         plan.setStatus(ClusterDrsPlan.Status.IN_PROGRESS);
         drsPlanDao.update(plan.getId(), plan);
 
+        List<VirtualMachine> dispatched = new ArrayList<>();
+
         for (ClusterDrsPlanMigrationVO migration : planMigrations) {
             try {
-                VirtualMachine vm = vmInstanceDao.findById(migration.getVmId());
+                VMInstanceVO vm = vmInstanceDao.findById(migration.getVmId());
                 Host host = hostDao.findById(migration.getDestHostId());
                 if (vm == null || host == null) {
                     throw new CloudRuntimeException(String.format("vm %s or host %s is not found", migration.getVmId(),
                             migration.getDestHostId()));
                 }
 
+                if (destinationViolatesAffinity(vm, host, dispatched)) {
+                    logger.warn("Skipping DRS migration of vm {} to host {}: it no longer satisfies the affinity " +
+                            "rules for that VM. The plan was generated against older state.", vm, host);
+                    migration.setStatus(JobInfo.Status.FAILED);
+                    drsPlanMigrationDao.update(migration.getId(), migration);
+                    continue;
+                }
+
                 logger.debug("Executing DRS plan {} for vm {} to host {}", plan, vm, host);
                 long jobId = createMigrateVMAsyncJob(vm, host, plan.getEventId());
                 AsyncJobVO job = asyncJobManager.getAsyncJob(jobId);
                 migration.setJobId(jobId);
                 migration.setStatus(job.getStatus());
                 drsPlanMigrationDao.update(migration.getId(), migration);
+
+                // the migration job has only been queued, so the database still shows the old host.
+                // record where it is headed so later migrations in this plan see it.
+                vm.setHostId(host.getId());
+                dispatched.add(vm);
             } catch (Exception e) {
                 logger.warn("Unable to execute DRS plan {} due to {}", plan, e.getMessage());
                 migration.setStatus(JobInfo.Status.FAILED);

server/src/test/java/org/apache/cloudstack/cluster/ClusterDrsServiceImplTest.java

@@ -29,12 +29,14 @@
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.host.Host;
 import com.cloud.host.HostVO;
+import com.cloud.deploy.DeploymentPlanner.ExcludeList;
 import com.cloud.host.dao.HostDao;
 import com.cloud.offering.ServiceOffering;
 import com.cloud.org.Cluster;
 import com.cloud.org.Grouping;
 import com.cloud.server.ManagementServer;
 import com.cloud.service.ServiceOfferingVO;
+import org.apache.cloudstack.affinity.AffinityGroupVMMapVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.utils.Pair;
 import com.cloud.utils.Ternary;
@@ -70,13 +72,15 @@
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Collections;
+import org.apache.cloudstack.jobs.JobInfo;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertNull;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -951,4 +955,86 @@ public void testProcessPlans() {
 
         Mockito.verify(clusterDrsService, Mockito.times(2)).executeDrsPlan(Mockito.any(ClusterDrsPlanVO.class));
     }
+
+    @Test
+    public void testDestinationViolatesAffinityWhenVmHasNoAffinityGroups() {
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(1L);
+        Mockito.when(affinityGroupVMMapDao.listByInstanceId(1L)).thenReturn(Collections.emptyList());
+
+        HostVO destHost = Mockito.mock(HostVO.class);
+
+        assertFalse(clusterDrsService.destinationViolatesAffinity(vm, destHost, Collections.emptyList()));
+        Mockito.verify(managementServer, Mockito.never())
+                .applyAffinityConstraints(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
+    }
+
+    @Test
+    public void testDestinationViolatesAffinityWhenDestHostIsExcluded() {
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(1L);
+        Mockito.when(vm.getServiceOfferingId()).thenReturn(5L);
+        Mockito.when(affinityGroupVMMapDao.listByInstanceId(1L))
+                .thenReturn(Collections.singletonList(Mockito.mock(AffinityGroupVMMapVO.class)));
+        Mockito.when(serviceOfferingDao.findByIdIncludingRemoved(1L, 5L))
+                .thenReturn(Mockito.mock(ServiceOfferingVO.class));
+
+        HostVO destHost = Mockito.mock(HostVO.class);
+        Mockito.when(destHost.getId()).thenReturn(20L);
+
+        ExcludeList excludes = new ExcludeList();
+        excludes.addHost(20L);
+        Mockito.when(managementServer.applyAffinityConstraints(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()))
+                .thenReturn(excludes);
+
+        assertTrue(clusterDrsService.destinationViolatesAffinity(vm, destHost, Collections.emptyList()));
+    }
+
+    @Test
+    public void testDestinationViolatesAffinityWhenDestHostIsAllowed() {
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vm.getId()).thenReturn(1L);
+        Mockito.when(vm.getServiceOfferingId()).thenReturn(5L);
+        Mockito.when(affinityGroupVMMapDao.listByInstanceId(1L))
+                .thenReturn(Collections.singletonList(Mockito.mock(AffinityGroupVMMapVO.class)));
+        Mockito.when(serviceOfferingDao.findByIdIncludingRemoved(1L, 5L))
+                .thenReturn(Mockito.mock(ServiceOfferingVO.class));
+
+        HostVO destHost = Mockito.mock(HostVO.class);
+        Mockito.when(destHost.getId()).thenReturn(20L);
+
+        ExcludeList excludes = new ExcludeList();
+        excludes.addHost(21L);
+        Mockito.when(managementServer.applyAffinityConstraints(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()))
+                .thenReturn(excludes);
+
+        assertFalse(clusterDrsService.destinationViolatesAffinity(vm, destHost, Collections.emptyList()));
+    }
+
+    @Test
+    public void testExecuteDrsPlanSkipsMigrationThatViolatesAffinity() {
+        ClusterDrsPlanVO plan = Mockito.mock(ClusterDrsPlanVO.class);
+        Mockito.when(plan.getId()).thenReturn(1L);
+
+        ClusterDrsPlanMigrationVO migration = Mockito.mock(ClusterDrsPlanMigrationVO.class);
+        Mockito.when(migration.getId()).thenReturn(7L);
+        Mockito.when(migration.getVmId()).thenReturn(1L);
+        Mockito.when(migration.getDestHostId()).thenReturn(20L);
+        Mockito.when(drsPlanMigrationDao.listPlanMigrationsToExecute(1L))
+                .thenReturn(Collections.singletonList(migration));
+
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        HostVO destHost = Mockito.mock(HostVO.class);
+        Mockito.when(vmInstanceDao.findById(1L)).thenReturn(vm);
+        Mockito.when(hostDao.findById(20L)).thenReturn(destHost);
+
+        Mockito.doReturn(true).when(clusterDrsService)
+                .destinationViolatesAffinity(Mockito.any(), Mockito.any(), Mockito.any());
+
+        clusterDrsService.executeDrsPlan(plan);
+
+        Mockito.verify(migration).setStatus(JobInfo.Status.FAILED);
+        Mockito.verify(clusterDrsService, Mockito.never())
+                .createMigrateVMAsyncJob(Mockito.any(), Mockito.any(), Mockito.anyLong());
+    }
 }