Add 'balancedweighted' VM allocation algorithm

PendingReleaseNotes

@@ -39,3 +39,22 @@ example.ver.1 > example.ver.2:
     which can now be attached to Instances. This is to prevent the Secondary
     Storage to grow to enormous sizes as Linux Distributions keep growing in
     size while a stripped down Linux should fit on a 2.88MB floppy.
+
+  * 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.
+
+    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.

api/src/main/java/com/cloud/deploy/DeploymentClusterPlanner.java

@@ -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

api/src/main/java/com/cloud/deploy/DeploymentPlanner.java

@@ -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 {

api/src/main/java/com/cloud/host/HostScoringWeights.java

@@ -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);
+}

engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDao.java

@@ -131,6 +131,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}. Every host in scope appears,
+     *         including hosts with no VMs.
+     */
+    Map<Long, Pair<Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter);
+
     Long countRunningAndStartingByAccount(long accountId);
 
     Long countByZoneAndState(long zoneId, State state);

engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDaoImpl.java

@@ -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;
@@ -154,6 +155,15 @@ 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.
+    private static final String COUNT_VMS_BY_HOST = "SELECT host.id, COUNT(vm.id), SUM(IF(%s, 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 " +
@@ -795,6 +805,44 @@ public Pair<List<Long>, Map<Long, Double>> listPodIdsInZoneByVmCount(long dataCe
         }
     }
 
+
+    @Override
+    public Map<Long, Pair<Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter) {
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
+        Map<Long, Pair<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()) {
+                result.put(rs.getLong(1), new Pair<>(rs.getLong(2), rs.getLong(3)));
+            }
+            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();

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

@@ -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);

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

@@ -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);
+    }
+}