Fix three defects found in review of the weighted scoring

Allocated fraction was measured against the wrong total. op_host_capacity
stores totals raw and overprovisioning is applied when they are read, so
dividing by the stored total made the fraction reach 1 at the host's
physical size. On a cluster overcommitted 10 times every host clamped to
1, killing both the allocation term and the dominant resource term - on
exactly the clusters this algorithm is for.

- apply the cluster ratio to the denominator
- drop hosts missing a CPU or memory capacity row instead of scoring the
  missing resource as untouched, which made them rank first

Utilisation thresholds could be bypassed. Held-back hosts were appended
before the random spread was applied, so the spread could shuffle a busy
host into the lead. With 1 healthy host and a spread of 3, two thirds of
deployments picked an over-threshold host.

- spread over healthy hosts only, before anything else is appended

A host with no load samples was treated as idle. It was exempt from the
thresholds and its dominant resource term fell back to allocation, so a
host with broken stats outranked every measured host and collected the
deployments.

- rank hosts we cannot measure behind every host we can
- when nothing can be measured, ranking falls back to allocation as before

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

Author: bhouse-nexthop

server/src/main/java/com/cloud/agent/manager/allocator/impl/WeightedHostScorer.java

@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -34,9 +35,12 @@
 import com.cloud.capacity.CapacityManager;
 import com.cloud.capacity.CapacityVO;
 import com.cloud.capacity.dao.CapacityDao;
+import com.cloud.dc.ClusterDetailsDao;
+import com.cloud.dc.ClusterDetailsVO;
 import com.cloud.host.Host;
 import com.cloud.host.HostScoringWeights;
 import com.cloud.utils.component.AdapterBase;
+import com.cloud.vm.VmDetailConstants;
 import com.cloud.vm.dao.VMInstanceDao;
 
 /**
@@ -109,6 +113,9 @@ public class WeightedHostScorer extends AdapterBase implements Configurable {
     @Inject
     private CapacityDao capacityDao;
 
+    @Inject
+    private ClusterDetailsDao clusterDetailsDao;
+
     @Inject
     private VMInstanceDao vmInstanceDao;
 
@@ -128,19 +135,53 @@ public List<Host> rank(long zoneId, Long podId, Long clusterId, List<? extends H
 
         Map<Long, Double> scores = score(zoneId, podId, clusterId, hosts);
 
-        List<Host> scored = hosts.stream().filter(h -> scores.containsKey(h.getId())).collect(Collectors.toList());
-        List<Host> unscored = hosts.stream().filter(h -> !scores.containsKey(h.getId())).collect(Collectors.toList());
-        scored.sort((a, b) -> Double.compare(scores.get(a.getId()), scores.get(b.getId())));
+        List<Host> unscored = new ArrayList<>();
+        List<Host> measured = new ArrayList<>();
+        List<Host> unmeasured = new ArrayList<>();
+        for (Host host : hosts) {
+            if (!scores.containsKey(host.getId())) {
+                unscored.add(host);
+            } else if (hostLoadTracker.getLoad(host.getId()).isUsable()) {
+                measured.add(host);
+            } else {
+                unmeasured.add(host);
+            }
+        }
+
+        Comparator<Host> byScore = Comparator.comparingDouble(h -> scores.get(h.getId()));
+        measured.sort(byScore);
+        unmeasured.sort(byScore);
 
-        List<Host> admitted = applyUtilisationThresholds(clusterId, scored);
-        applySelectionSpread(clusterId, admitted);
+        List<Host> healthy = new ArrayList<>();
+        List<Host> tooBusy = new ArrayList<>();
+        partitionByUtilisation(clusterId, measured, healthy, tooBusy);
 
-        logger.debug("Weighted host ranking: {}", () -> admitted.stream()
+        List<Host> result = new ArrayList<>();
+        if (healthy.isEmpty() && unmeasured.isEmpty()) {
+            logger.warn("Every candidate host is above its utilisation threshold, so the thresholds are being "
+                    + "ignored for this deployment. The cluster is short of capacity.");
+            result.addAll(tooBusy);
+            applySelectionSpread(clusterId, result);
+        } else {
+            result.addAll(healthy);
+            // spread only over hosts known to be healthy, before anything else is appended,
+            // otherwise a busy or unmeasured host can be shuffled into the lead
+            applySelectionSpread(clusterId, result);
+            // a host we cannot measure is not assumed to be idle: it ranks behind every host we can
+            result.addAll(unmeasured);
+            result.addAll(tooBusy);
+        }
+
+        if (!tooBusy.isEmpty()) {
+            logger.debug("Holding back {} host(s) above their utilisation threshold: {}", tooBusy.size(), tooBusy);
+        }
+        logger.debug("Weighted host ranking: {}", () -> result.stream()
+                .filter(h -> scores.containsKey(h.getId()))
                 .map(h -> String.format("%s=%.4f", h.getName(), scores.get(h.getId())))
                 .collect(Collectors.joining(", ")));
 
-        admitted.addAll(unscored);
-        return admitted;
+        result.addAll(unscored);
+        return result;
     }
 
     protected Map<Long, Double> score(long zoneId, Long podId, Long clusterId, List<? extends Host> hosts) {
@@ -165,27 +206,58 @@ protected Map<Long, Double> score(long zoneId, Long podId, Long clusterId, List<
     }
 
     /**
-     * Allocated CPU and memory as a fraction of what the host advertises after overprovisioning,
-     * which is the same basis the existing allocators use.
+     * Allocated CPU and memory as a fraction of what a host can hand out.
+     *
+     * op_host_capacity stores totals raw; overprovisioning is applied when they are read, so the
+     * cluster's ratio has to be applied here too. Without it the fraction reaches 1 at the host's
+     * physical size and every host on an overcommitted cluster clamps to 1, which is where this
+     * algorithm is most needed.
+     *
+     * Only hosts with both a CPU and a memory row are returned. A host missing one would otherwise
+     * score as if that resource were untouched, making it the most attractive host in the cluster.
      */
     protected Map<Long, Double[]> allocatedFractions(List<CapacityVO> capacities) {
         Map<Long, Double[]> fractions = new HashMap<>();
+        Map<Long, Integer> seen = new HashMap<>();
         for (CapacityVO capacity : capacities) {
             long total = capacity.getTotalCapacity();
             if (total <= 0) {
                 continue;
             }
-            double used = (double) (capacity.getUsedCapacity() + capacity.getReservedCapacity()) / total;
+            boolean isCpu = capacity.getCapacityType() == Capacity.CAPACITY_TYPE_CPU;
+            float overcommit = overcommitRatio(capacity.getClusterId(), isCpu);
+            double allocatable = total * overcommit;
+            double used = (double) (capacity.getUsedCapacity() + capacity.getReservedCapacity()) / allocatable;
+
             Double[] entry = fractions.computeIfAbsent(capacity.getHostOrPoolId(), id -> new Double[] {0.0, 0.0});
-            if (capacity.getCapacityType() == Capacity.CAPACITY_TYPE_CPU) {
-                entry[0] = clamp(used);
-            } else {
-                entry[1] = clamp(used);
-            }
+            entry[isCpu ? 0 : 1] = clamp(used);
+            seen.merge(capacity.getHostOrPoolId(), isCpu ? 1 : 2, Integer::sum);
         }
+        fractions.keySet().removeIf(hostId -> seen.getOrDefault(hostId, 0) != 3);
         return fractions;
     }
 
+    /**
+     * The cluster's overprovisioning factor, defaulting to none if it cannot be read.
+     */
+    protected float overcommitRatio(Long clusterId, boolean forCpu) {
+        if (clusterId == null) {
+            return 1f;
+        }
+        String key = forCpu ? VmDetailConstants.CPU_OVER_COMMIT_RATIO : VmDetailConstants.MEMORY_OVER_COMMIT_RATIO;
+        ClusterDetailsVO detail = clusterDetailsDao.findDetail(clusterId, key);
+        if (detail == null || detail.getValue() == null) {
+            return 1f;
+        }
+        try {
+            float ratio = Float.parseFloat(detail.getValue());
+            return ratio > 0 ? ratio : 1f;
+        } catch (NumberFormatException e) {
+            logger.warn("Cluster {} has an unreadable {} of [{}], treating it as 1.", clusterId, key, detail.getValue());
+            return 1f;
+        }
+    }
+
     /**
      * The blend. Every term is a fraction of the host's capacity for that resource so the weights
      * are directly comparable, and the dominant resource term is added on top of the weighted mean
@@ -238,35 +310,21 @@ protected double dominantResource(double cpuAllocated, double memoryAllocated, H
     }
 
     /**
-     * Holds back hosts that are measurably too busy, unless that would leave nothing to deploy on,
-     * in which case ranking alone decides and the caller's capacity checks still apply.
+     * Splits measurably busy hosts out from the rest. Only hosts with load samples can be held
+     * back; a host that cannot be measured is dealt with by the caller.
      */
-    protected List<Host> applyUtilisationThresholds(Long clusterId, List<Host> ranked) {
+    protected void partitionByUtilisation(Long clusterId, List<Host> measured, List<Host> healthy, List<Host> tooBusy) {
         double cpuThreshold = valueIn(CpuUtilisationThreshold, clusterId);
         double memoryThreshold = valueIn(MemoryUtilisationThreshold, clusterId);
 
-        List<Host> admitted = new ArrayList<>();
-        List<Host> heldBack = new ArrayList<>();
-        for (Host host : ranked) {
+        for (Host host : measured) {
             HostLoad load = hostLoadTracker.getLoad(host.getId());
-            if (load.isUsable()
-                    && (load.getCpuUtilisation() > cpuThreshold || load.getMemoryUtilisation() > memoryThreshold)) {
-                heldBack.add(host);
+            if (load.getCpuUtilisation() > cpuThreshold || load.getMemoryUtilisation() > memoryThreshold) {
+                tooBusy.add(host);
             } else {
-                admitted.add(host);
+                healthy.add(host);
             }
         }
-
-        if (admitted.isEmpty()) {
-            logger.warn("Every candidate host is above its utilisation threshold, so the thresholds are being "
-                    + "ignored for this deployment. The cluster is short of capacity.");
-            return heldBack;
-        }
-        if (!heldBack.isEmpty()) {
-            logger.debug("Holding back {} host(s) above their utilisation threshold: {}", heldBack.size(), heldBack);
-            admitted.addAll(heldBack);
-        }
-        return admitted;
     }
 
     /**

server/src/test/java/com/cloud/agent/manager/allocator/impl/WeightedHostScorerTest.java

@@ -130,10 +130,13 @@ public void testBusyHostIsHeldBackByThreshold() {
         Mockito.when(hostLoadTracker.getLoad(quiet.getId())).thenReturn(new HostLoad(0.10, 0.10, 10));
         Mockito.when(hostLoadTracker.getLoad(busy.getId())).thenReturn(new HostLoad(0.99, 0.10, 10));
 
-        List<Host> result = scorer.applyUtilisationThresholds(null, new ArrayList<>(List.of(busy, quiet)));
+        List<Host> healthy = new ArrayList<>();
+        List<Host> tooBusy = new ArrayList<>();
+        scorer.partitionByUtilisation(null, new ArrayList<>(List.of(quiet, busy)), healthy, tooBusy);
 
-        assertSame("host over the CPU threshold must fall behind", quiet, result.get(0));
-        assertSame(busy, result.get(1));
+        assertEquals(1, healthy.size());
+        assertSame(quiet, healthy.get(0));
+        assertSame("host over the CPU threshold must be held back", busy, tooBusy.get(0));
     }
 
     @Test
@@ -142,9 +145,12 @@ public void testThresholdIsIgnoredWhenEveryHostIsBusy() {
         Host b = host("b");
         Mockito.when(hostLoadTracker.getLoad(Mockito.anyLong())).thenReturn(new HostLoad(0.99, 0.99, 10));
 
-        List<Host> result = scorer.applyUtilisationThresholds(null, new ArrayList<>(List.of(a, b)));
+        List<Host> healthy = new ArrayList<>();
+        List<Host> tooBusy = new ArrayList<>();
+        scorer.partitionByUtilisation(null, new ArrayList<>(List.of(a, b)), healthy, tooBusy);
 
-        assertEquals("deployment must still be possible when the whole cluster is busy", 2, result.size());
+        assertEquals("both hosts are over threshold", 2, tooBusy.size());
+        assertTrue(healthy.isEmpty());
     }
 
     @Test