@@ -91,3 +91,31 @@ example.ver.1 > example.ver.2:
     no device-mapper multipath configuration is required. A new
     Storage.StoragePoolType.NVMeTCP + MultipathNVMeOFAdapterBase /
     NVMeTCPAdapter on the KVM side back the new pool type.
+
+  * New VM allocation algorithm 'balancedweighted' for vm.allocation.algorithm. Existing
+    algorithms and the default are unchanged; the new one is opt-in.
+
+    Existing algorithms rank hosts on allocated capacity alone. Under a large overprovisioning
+    factor that reads badly: allocation is measured against a total that has already been
+    multiplied by the factor, so a host under real strain can still report a low percentage and
+    keep attracting new VMs. Concurrent deployments make it worse, since they all read the same
+    figures before any of them is accounted for.
+
+    'balancedweighted' ranks hosts on a blend of allocated CPU and memory, measured CPU and
+    memory utilisation, VM count, and how many VMs started on the host recently. It holds back
+    hosts that are measurably too busy, and chooses at random from among the best scoring hosts
+    so that simultaneous deployments do not all pick the same one.
+
+    It also holds back hosts that already have host.weighted.starting.vms.threshold VMs sitting
+    in the Starting state, so that a burst of deployments does not queue up behind itself on one
+    host. This is the only signal that reacts within a single deployment: capacity is not charged
+    until a VM reaches Starting and a utilisation average is a sample interval behind, so neither
+    notices a burst until it is over. Set the threshold to 0 to disable it.
+
+    Held-back hosts are never excluded, only moved to the back of the list, so no threshold can
+    turn a deployment that would have succeeded into an InsufficientServerCapacity failure.
+
+    Tuned with the host.weighted.* settings, most of which are cluster scoped. Measured
+    utilisation comes from a moving average configured with host.load.sample.interval and
+    host.load.half.life. Until a management server has collected samples, ranking falls back to
+    allocation figures alone.
@@ -62,11 +62,14 @@ public interface DeploymentClusterPlanner extends DeploymentPlanner {
             "vm.allocation.algorithm",
             "Advanced",
             "random",
-            "Order in which hosts within a cluster will be considered for VM allocation. The value can be 'random', 'firstfit', 'userdispersing', or 'firstfitleastconsumed'.",
+            "Order in which hosts within a cluster will be considered for VM allocation. The value can be 'random', "
+                    + "'firstfit', 'userdispersing', 'firstfitleastconsumed', or 'balancedweighted'. 'balancedweighted' "
+                    + "ranks hosts on a blend of allocated capacity, measured utilisation, VM count and how many VMs "
+                    + "started recently, and is tuned with the host.weighted.* settings.",
             true,
             ConfigKey.Scope.Global, null, null, null, null, null,
             ConfigKey.Kind.Select,
-            "random,firstfit,userdispersing,firstfitleastconsumed");
+            "random,firstfit,userdispersing,firstfitleastconsumed,balancedweighted");
 
     /**
      * This is called to determine list of possible clusters where a virtual
 
@@ -70,7 +70,7 @@ public interface DeploymentPlanner extends Adapter {
     boolean canHandle(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid);
 
     public enum AllocationAlgorithm {
-        random, firstfit, userdispersing, firstfitleastconsumed;
+        random, firstfit, userdispersing, firstfitleastconsumed, balancedweighted;
     }
 
     public enum PlannerResourceUsage {
 
@@ -0,0 +1,57 @@
+// 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.host;
+
+import org.apache.cloudstack.framework.config.ConfigKey;
+
+/**
+ * How much each signal counts when ranking hosts by how loaded they are.
+ *
+ * Shared by initial placement and by rebalancing on purpose. If the two weighted these differently
+ * they would disagree about which host is the better one, and rebalancing could move VMs off hosts
+ * that placement had just chosen, only for placement to put them back.
+ *
+ * Weights are relative to each other; only their ratios matter, and zero disables a term. Terms that
+ * only make sense for one of the two - how many VMs a host carries, how many started recently - stay
+ * with whichever uses them.
+ */
+public interface HostScoringWeights {
+
+    String WEIGHT_DESCRIPTION_SUFFIX = " Relative weight, only meaningful compared with the other "
+            + "host.weighted.* weights. Zero disables the term. Used by both the 'balancedweighted' "
+            + "allocation algorithm and the 'weighted' DRS algorithm.";
+
+    ConfigKey<Double> CpuAllocatedWeight = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
+            Double.class, "host.weighted.cpu.allocated.weight", "1.0",
+            "How much CPU allocated on a host counts against it." + WEIGHT_DESCRIPTION_SUFFIX,
+            true, ConfigKey.Scope.Cluster);
+
+    ConfigKey<Double> CpuUsedWeight = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
+            Double.class, "host.weighted.cpu.used.weight", "2.0",
+            "How much measured CPU utilisation counts against a host." + WEIGHT_DESCRIPTION_SUFFIX,
+            true, ConfigKey.Scope.Cluster);
+
+    ConfigKey<Double> MemoryAllocatedWeight = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
+            Double.class, "host.weighted.memory.allocated.weight", "1.0",
+            "How much memory allocated on a host counts against it." + WEIGHT_DESCRIPTION_SUFFIX,
+            true, ConfigKey.Scope.Cluster);
+
+    ConfigKey<Double> MemoryUsedWeight = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
+            Double.class, "host.weighted.memory.used.weight", "2.0",
+            "How much measured memory utilisation counts against a host." + WEIGHT_DESCRIPTION_SUFFIX,
+            true, ConfigKey.Scope.Cluster);
+}
@@ -24,6 +24,7 @@
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -935,7 +936,14 @@ public List<HostVO> listAllUpAndEnabledNonHAHosts(Type type, Long clusterId, Lon
         sc.setParameters("status", Status.Up);
         sc.setParameters("resourceState", ResourceState.Enabled);
 
-        return listBy(sc);
+        // The tag join carries no DISTINCT, so a host comes back once per non-rule tag it holds.
+        // Callers read this as a set of candidate hosts, so collapse the repeats here rather than
+        // leave each of them to cope with the same host arriving more than once.
+        Map<Long, HostVO> distinctHosts = new LinkedHashMap<>();
+        for (HostVO host : listBy(sc)) {
+            distinctHosts.putIfAbsent(host.getId(), host);
+        }
+        return new ArrayList<>(distinctHosts.values());
     }
 
     @Override
 
@@ -25,6 +25,7 @@
 
 import com.cloud.hypervisor.Hypervisor;
 import com.cloud.utils.Pair;
+import com.cloud.utils.Ternary;
 import com.cloud.utils.db.GenericDao;
 import com.cloud.utils.fsm.StateDao;
 import com.cloud.vm.VMInstanceVO;
@@ -131,6 +132,18 @@ public interface VMInstanceDao extends GenericDao<VMInstanceVO, Long>, StateDao<
 
     List<Long> listHostIdsByVmCount(long dcId, Long podId, Long clusterId, long accountId);
 
+    /**
+     * Counts the VMs occupying each host in a zone, pod or cluster, in one query.
+     *
+     * @param changedStateAfter
+     *         cut-off for the second count: VMs whose state last changed after this. Approximates
+     *         the VMs still working through their startup load, which neither allocation nor a
+     *         utilisation average has caught up with yet.
+     * @return host id to {total VMs, VMs that changed state recently, VMs still in Starting}. Every
+     *         host in scope appears, including hosts with no VMs.
+     */
+    Map<Long, Ternary<Long, Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter);
+
     Long countRunningAndStartingByAccount(long accountId);
 
     Long countByZoneAndState(long zoneId, State state);
 
@@ -18,6 +18,7 @@
 
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
+import java.sql.Timestamp;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -45,6 +46,7 @@
 import com.cloud.tags.dao.ResourceTagDao;
 import com.cloud.utils.DateUtil;
 import com.cloud.utils.Pair;
+import com.cloud.utils.Ternary;
 import com.cloud.utils.db.Attribute;
 import com.cloud.utils.db.DB;
 import com.cloud.utils.db.Filter;
@@ -156,6 +158,21 @@ public class VMInstanceDaoImpl extends GenericDaoBase<VMInstanceVO, Long> implem
     private static final String COUNT_VMS_BASED_ON_VGPU_TYPES2 =
             "GROUP BY gpu_card.name, vgpu_profile.name";
 
+    // %s is the "changed state recently" test, or a constant 0 when no cut-off is given. It is not
+    // a bound parameter because there is no timestamp that reliably means "never" - update_time is
+    // a TIMESTAMP column, so anything past 2038 is out of range.
+    //
+    // The third count is VMs still in Starting. Unlike the recent-state-change count it does not
+    // expire on a clock: a VM that has been starting for ten minutes still counts, which is the
+    // case that matters, since a host that is struggling is exactly the one whose VMs take longest
+    // to leave Starting.
+    private static final String COUNT_VMS_BY_HOST = "SELECT host.id, COUNT(vm.id), SUM(IF(%s, 1, 0)), " +
+            "SUM(IF(vm.state = 'Starting', 1, 0)) " +
+            "FROM `cloud`.`host` host LEFT JOIN `cloud`.`vm_instance` vm " +
+            "ON vm.host_id = host.id AND vm.state IN ('Running', 'Starting', 'Stopping', 'Migrating') " +
+            "AND vm.removed IS NULL WHERE host.type = 'Routing' AND host.removed IS NULL AND host.data_center_id = ? ";
+    private static final String COUNT_VMS_BY_HOST_PART2 = " GROUP BY host.id ";
+
     private static final String UPDATE_SYSTEM_VM_TEMPLATE_ID_FOR_HYPERVISOR = "UPDATE `cloud`.`vm_instance` SET vm_template_id = ? WHERE type <> 'User' AND hypervisor_type = ? AND removed is NULL";
 
     private static final String COUNT_VMS_BY_ZONE_AND_STATE_AND_HOST_TAG = "SELECT COUNT(1) FROM vm_instance vi JOIN service_offering so ON vi.service_offering_id=so.id " +
@@ -801,6 +818,45 @@ public Pair<List<Long>, Map<Long, Double>> listPodIdsInZoneByVmCount(long dataCe
         }
     }
 
+
+    @Override
+    public Map<Long, Ternary<Long, Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter) {
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
+        Map<Long, Ternary<Long, Long, Long>> result = new HashMap<>();
+        String sql = String.format(COUNT_VMS_BY_HOST, changedStateAfter != null ? "vm.update_time > ?" : "0");
+        if (podId != null) {
+            sql = sql + " AND host.pod_id = ? ";
+        }
+        if (clusterId != null) {
+            sql = sql + " AND host.cluster_id = ? ";
+        }
+        sql = sql + COUNT_VMS_BY_HOST_PART2;
+        try {
+            PreparedStatement pstmt = txn.prepareAutoCloseStatement(sql);
+            int index = 1;
+            if (changedStateAfter != null) {
+                pstmt.setTimestamp(index++, new Timestamp(changedStateAfter.getTime()));
+            }
+            pstmt.setLong(index++, dcId);
+            if (podId != null) {
+                pstmt.setLong(index++, podId);
+            }
+            if (clusterId != null) {
+                pstmt.setLong(index++, clusterId);
+            }
+            ResultSet rs = pstmt.executeQuery();
+            while (rs.next()) {
+                // SUM over no rows is NULL, which getLong reports as 0 - correct for an empty host
+                result.put(rs.getLong(1), new Ternary<>(rs.getLong(2), rs.getLong(3), rs.getLong(4)));
+            }
+            return result;
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("DB Exception on: " + sql, e);
+        } catch (Exception e) {
+            throw new CloudRuntimeException("Caught: " + sql, e);
+        }
+    }
+
     @Override
     public List<Long> listHostIdsByVmCount(long dcId, Long podId, Long clusterId, long accountId) {
         TransactionLegacy txn = TransactionLegacy.currentTxn();
 
@@ -39,6 +39,7 @@
 
 import com.cloud.cpu.CPU;
 import com.cloud.host.Host;
+import com.cloud.host.HostTagVO;
 import com.cloud.host.HostVO;
 import com.cloud.host.Status;
 import com.cloud.hypervisor.Hypervisor;
@@ -60,6 +61,32 @@ public class HostDaoImplTest {
     @Mock
     private SearchCriteria<HostVO> mockSearchCriteria;
 
+    @Test
+    public void testListAllUpAndEnabledNonHAHostsCollapsesRepeatedHosts() {
+        // the host_tags join carries no DISTINCT, so a host comes back once per non-rule tag it
+        // holds. Callers read the result as a set of candidate hosts.
+        SearchBuilder<HostTagVO> tagSearchBuilder = mock(SearchBuilder.class);
+        when(tagSearchBuilder.entity()).thenReturn(mock(HostTagVO.class));
+        HostTagsDao hostTagsDao = mock(HostTagsDao.class);
+        when(hostTagsDao.createSearchBuilder()).thenReturn(tagSearchBuilder);
+        hostDao._hostTagsDao = hostTagsDao;
+
+        when(mockSearchBuilder.entity()).thenReturn(mock(HostVO.class));
+        when(mockSearchBuilder.create()).thenReturn(mockSearchCriteria);
+        doReturn(mockSearchBuilder).when(hostDao).createSearchBuilder();
+
+        HostVO twoTags = mock(HostVO.class);
+        when(twoTags.getId()).thenReturn(1L);
+        HostVO oneTag = mock(HostVO.class);
+        when(oneTag.getId()).thenReturn(2L);
+        doReturn(Arrays.asList(twoTags, twoTags, oneTag)).when(hostDao).listBy(any(SearchCriteria.class));
+
+        List<HostVO> hosts = hostDao.listAllUpAndEnabledNonHAHosts(Host.Type.Routing, 3L, 2L, 1L, null);
+
+        assertEquals(2, hosts.size());
+        assertEquals(Arrays.asList(1L, 2L), hosts.stream().map(HostVO::getId).collect(Collectors.toList()));
+    }
+
     @Test
     public void testCountUpAndEnabledHostsInZone() {
         long testZoneId = 100L;
 
@@ -16,6 +16,7 @@
 // under the License.
 package com.cloud.agent.manager.allocator.impl;
 
+import static com.cloud.deploy.DeploymentPlanner.AllocationAlgorithm.balancedweighted;
 import static com.cloud.deploy.DeploymentPlanner.AllocationAlgorithm.firstfitleastconsumed;
 import static com.cloud.deploy.DeploymentPlanner.AllocationAlgorithm.random;
 import static com.cloud.deploy.DeploymentPlanner.AllocationAlgorithm.userdispersing;
@@ -95,6 +96,8 @@ public class FirstFitAllocator extends BaseAllocator {
     CapacityDao _capacityDao;
     @Inject
     VMInstanceDetailsDao vmInstanceDetailsDao;
+    @Inject
+    WeightedHostScorer weightedHostScorer;
 
     boolean _checkHvm = true;
 
@@ -209,6 +212,8 @@ protected List<Host> allocateTo(VirtualMachineProfile vmProfile, DeploymentPlan
             hosts = reorderHostsByNumberOfVms(plan, hosts, account);
         } else if (firstfitleastconsumed.toString().equals(vmAllocationAlgorithm)) {
             hosts = reorderHostsByCapacity(plan, hosts);
+        } else if (balancedweighted.toString().equals(vmAllocationAlgorithm)) {
+            hosts = weightedHostScorer.rank(plan.getDataCenterId(), plan.getPodId(), plan.getClusterId(), hosts);
         }
 
         logger.debug("FirstFitAllocator has {} hosts to check for allocation {}.", hosts.size(), hosts);
 
@@ -0,0 +1,60 @@
+// 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.agent.manager.allocator.impl;
+
+/**
+ * Smoothed view of what a host is actually doing, as opposed to what has been allocated on it.
+ * Fractions are of the host's real capacity and ignore overprovisioning.
+ */
+public class HostLoad {
+
+    public static final HostLoad UNKNOWN = new HostLoad(0, 0, 0);
+
+    private final double cpuUtilisation;
+    private final double memoryUtilisation;
+    private final long samples;
+
+    public HostLoad(double cpuUtilisation, double memoryUtilisation, long samples) {
+        this.cpuUtilisation = cpuUtilisation;
+        this.memoryUtilisation = memoryUtilisation;
+        this.samples = samples;
+    }
+
+    public double getCpuUtilisation() {
+        return cpuUtilisation;
+    }
+
+    public double getMemoryUtilisation() {
+        return memoryUtilisation;
+    }
+
+    public long getSamples() {
+        return samples;
+    }
+
+    /**
+     * False until enough has been observed to rank on. Callers fall back to allocation figures.
+     */
+    public boolean isUsable() {
+        return samples > 0;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("HostLoad[cpu=%.3f, memory=%.3f, samples=%d]", cpuUtilisation, memoryUtilisation, samples);
+    }
+}
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ public interface DeploymentPlanner extends Adapter {`
`70`	`70`	`boolean canHandle(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid);`
`71`	`71`
`72`	`72`	`public enum AllocationAlgorithm {`
`73`		`- random, firstfit, userdispersing, firstfitleastconsumed;`
	`73`	`+ random, firstfit, userdispersing, firstfitleastconsumed, balancedweighted;`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`public enum PlannerResourceUsage {`