Address the rest of the weighted scoring review

Utilisation average
  - a host whose agent stops reporting kept vouching for itself forever:
    StatsCollector hands back the previous entry when a poll fails, and
    that unchanged reading was folded again every minute. Detect the
    repeat, and expire an average that stops being updated
  - sample on a scheduled executor catching Throwable, not a Timer, which
    dies permanently and silently on one escaping error
  - only collect when an algorithm that reads the figures is selected
  - read the half life once per sample rather than inside the map update
  - document what getCpuUtilization means per hypervisor: it is what this
    assumes on KVM, a reservation figure on VMware, and scaled by core
    count on XenServer

Scoring
  - a negative weight would rank the most loaded host first; floor at
    zero and say so
  - zeroing all six terms no longer discards the dominant resource term
  - read weights once per ranking instead of once per host

Queries
  - one query per ranking instead of two, returning both counts
  - count Stopping VMs, which still hold their host
  - correct the doc: every host in scope is returned, including empty ones

Tests
  - cover rank() end to end, which is where the defects were: the
    capacity denominator, the thresholds, the spread and the ordering of
    measured against unmeasured hosts
  - the distribution simulation drew different random streams per arm, so
    the arms saw different workloads. Fix the workload up front and add
    an allocation-only-plus-spread control, which shows the scoring and
    not the spread is what evens out real load

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

Author: bhouse-nexthop

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

@@ -132,14 +132,16 @@ 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.
+     * Counts the VMs occupying each host in a zone, pod or cluster, in one query.
      *
-     * @param startedAfter
-     *         when set, counts only VMs that last changed state after this time, which approximates
-     *         the VMs that have recently started and are still working through their startup load
-     * @return host id to VM count, hosts with no VMs omitted
+     * @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, Long> countVmsByHost(long dcId, Long podId, Long clusterId, Date startedAfter);
+    Map<Long, Pair<Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter);
 
     Long countRunningAndStartingByAccount(long accountId);
 

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

@@ -155,9 +155,10 @@ 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";
 
-    private static final String COUNT_VMS_BY_HOST = "SELECT host.id, COUNT(vm.id) FROM `cloud`.`host` host " +
-            "LEFT JOIN `cloud`.`vm_instance` vm ON vm.host_id = host.id AND vm.state IN ('Running', 'Starting', 'Migrating') " +
-            "AND vm.removed IS NULL %s WHERE host.type = 'Routing' AND host.removed IS NULL AND host.data_center_id = ? ";
+    private static final String COUNT_VMS_BY_HOST = "SELECT host.id, COUNT(vm.id), SUM(IF(vm.update_time > ?, 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";
@@ -803,10 +804,10 @@ public Pair<List<Long>, Map<Long, Double>> listPodIdsInZoneByVmCount(long dataCe
 
 
     @Override
-    public Map<Long, Long> countVmsByHost(long dcId, Long podId, Long clusterId, Date startedAfter) {
+    public Map<Long, Pair<Long, Long>> countVmsByHost(long dcId, Long podId, Long clusterId, Date changedStateAfter) {
         TransactionLegacy txn = TransactionLegacy.currentTxn();
-        Map<Long, Long> result = new HashMap<>();
-        String sql = String.format(COUNT_VMS_BY_HOST, startedAfter != null ? " AND vm.update_time > ? " : "");
+        Map<Long, Pair<Long, Long>> result = new HashMap<>();
+        String sql = COUNT_VMS_BY_HOST;
         if (podId != null) {
             sql = sql + " AND host.pod_id = ? ";
         }
@@ -817,19 +818,19 @@ public Map<Long, Long> countVmsByHost(long dcId, Long podId, Long clusterId, Dat
         try {
             PreparedStatement pstmt = txn.prepareAutoCloseStatement(sql);
             int index = 1;
-            if (startedAfter != null) {
-                pstmt.setTimestamp(index++, new Timestamp(startedAfter.getTime()));
-            }
+            // a cut-off in the future counts nothing as recent, which is what a null asks for
+            long cutOff = changedStateAfter != null ? changedStateAfter.getTime() : Long.MAX_VALUE;
+            pstmt.setTimestamp(index++, new Timestamp(cutOff));
             pstmt.setLong(index++, dcId);
             if (podId != null) {
                 pstmt.setLong(index++, podId);
             }
             if (clusterId != null) {
-                pstmt.setLong(index, clusterId);
+                pstmt.setLong(index++, clusterId);
             }
             ResultSet rs = pstmt.executeQuery();
             while (rs.next()) {
-                result.put(rs.getLong(1), rs.getLong(2));
+                result.put(rs.getLong(1), new Pair<>(rs.getLong(2), rs.getLong(3)));
             }
             return result;
         } catch (SQLException e) {

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

@@ -17,22 +17,26 @@
 package com.cloud.agent.manager.allocator.impl;
 
 import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import javax.inject.Inject;
 
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.Configurable;
-import org.apache.cloudstack.managed.context.ManagedContextTimerTask;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
 
 import com.cloud.host.HostStats;
 import com.cloud.host.HostVO;
 import com.cloud.host.Status;
 import com.cloud.host.dao.HostDao;
 import com.cloud.server.StatsCollector;
+import com.cloud.deploy.DeploymentClusterPlanner;
+import com.cloud.deploy.DeploymentPlanner.AllocationAlgorithm;
 import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.concurrency.NamedThreadFactory;
 
 /**
  * Keeps a smoothed view of how hard each host is actually working.
@@ -45,6 +49,19 @@
  * The average is per management server and is not persisted. Every management server polls every
  * host, so all of them converge on the same picture, and a restarted server simply reports nothing
  * usable until it has sampled - callers then fall back to allocation figures.
+ *
+ * What getCpuUtilization means depends on the hypervisor, and only KVM reports what this class
+ * assumes:
+ *
+ * <ul>
+ *   <li>KVM reports busy time as a percentage of the host's cores, which is what is wanted.</li>
+ *   <li>VMware reports the share of CPU that is reserved rather than the share that is busy, so
+ *       the CPU term becomes a second allocation signal there rather than a load signal.</li>
+ *   <li>XenServer sums per-core averages without dividing by core count, so the value ranges up to
+ *       the number of cores and is under-reported here by roughly that factor.</li>
+ * </ul>
+ *
+ * Memory is taken as used over total and is sound everywhere.
  */
 public class HostLoadTracker extends ManagerBase implements Configurable {
 
@@ -54,6 +71,13 @@ public class HostLoadTracker extends ManagerBase implements Configurable {
                     "consider actual load. Should not be shorter than host.stats.interval.",
             false, ConfigKey.Scope.Global);
 
+    public static final ConfigKey<Integer> HostLoadStaleAfter = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
+            Integer.class, "host.load.stale.after", "600",
+            "Seconds after which a host's utilisation average is considered out of date and stops being used " +
+                    "for placement. A host whose agent stops reporting would otherwise keep vouching for itself " +
+                    "with figures that never change.",
+            true, ConfigKey.Scope.Global);
+
     public static final ConfigKey<Integer> HostLoadHalfLife = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED,
             Integer.class, "host.load.half.life", "300",
             "Half life in seconds of the moving average of host utilisation. Larger values react more " +
@@ -68,35 +92,41 @@ public class HostLoadTracker extends ManagerBase implements Configurable {
 
     private final Map<Long, Sample> samples = new ConcurrentHashMap<>();
 
-    private Timer timer;
+    private ScheduledExecutorService executor;
 
     @Override
     public boolean start() {
-        int interval = Math.max(1, HostLoadSampleInterval.value()) * 1000;
-        TimerTask task = new ManagedContextTimerTask() {
+        int interval = Math.max(1, HostLoadSampleInterval.value());
+        executor = Executors.newSingleThreadScheduledExecutor(
+                new NamedThreadFactory("HostLoadTracker"));
+        // catch Throwable: an escaping error would cancel all future runs, and the failure would be
+        // silent - placement would quietly go back to ranking on allocation alone
+        executor.scheduleWithFixedDelay(new ManagedContextRunnable() {
             @Override
             protected void runInContext() {
                 try {
                     sampleAllHosts();
-                } catch (Exception e) {
-                    logger.warn("Unable to sample host load", e);
+                } catch (Throwable t) {
+                    logger.warn("Unable to sample host load", t);
                 }
             }
-        };
-        timer = new Timer("HostLoadTracker");
-        timer.schedule(task, interval, interval);
+        }, interval, interval, TimeUnit.SECONDS);
         return true;
     }
 
     @Override
     public boolean stop() {
-        if (timer != null) {
-            timer.cancel();
+        if (executor != null) {
+            executor.shutdownNow();
         }
         return true;
     }
 
     protected void sampleAllHosts() {
+        if (!isInUse()) {
+            samples.clear();
+            return;
+        }
         for (HostVO host : hostDao.listByType(com.cloud.host.Host.Type.Routing)) {
             if (host.getStatus() != Status.Up) {
                 samples.remove(host.getId());
@@ -106,6 +136,14 @@ protected void sampleAllHosts() {
         }
     }
 
+    /**
+     * Only the placement algorithms that read these figures pay for collecting them.
+     */
+    protected boolean isInUse() {
+        return AllocationAlgorithm.balancedweighted.toString()
+                .equals(DeploymentClusterPlanner.VmAllocationAlgorithm.value());
+    }
+
     protected void record(long hostId, HostStats stats) {
         record(hostId, stats, System.currentTimeMillis());
     }
@@ -118,18 +156,40 @@ protected void record(long hostId, HostStats stats, long now) {
         if (totalMemory <= 0) {
             return;
         }
-        // getCpuUtilization is a percentage of the host's real cores
+
+        Sample previous = samples.get(hostId);
+        if (previous != null && previous.isSameReadingAs(stats)) {
+            // StatsCollector keeps the previous entry when a poll fails, so an unchanged object is
+            // a reading we have already folded, not a fresh measurement
+            return;
+        }
+
+        // getCpuUtilization is a percentage of the host's real cores. That holds for KVM; see the
+        // class javadoc for what it means on other hypervisors.
         double cpu = clamp(stats.getCpuUtilization() / 100.0);
         double memory = clamp((totalMemory - stats.getFreeMemoryKBs()) / totalMemory);
+        int halfLife = HostLoadHalfLife.value();
 
-        samples.compute(hostId, (id, previous) -> previous == null
-                ? new Sample(cpu, memory, now)
-                : previous.fold(cpu, memory, now, HostLoadHalfLife.value()));
+        samples.compute(hostId, (id, current) -> current == null
+                ? new Sample(cpu, memory, now, stats)
+                : current.fold(cpu, memory, now, halfLife, stats));
     }
 
     public HostLoad getLoad(long hostId) {
+        return getLoad(hostId, System.currentTimeMillis());
+    }
+
+    protected HostLoad getLoad(long hostId, long now) {
         Sample sample = samples.get(hostId);
-        return sample == null ? HostLoad.UNKNOWN : sample.toHostLoad();
+        if (sample == null) {
+            return HostLoad.UNKNOWN;
+        }
+        long staleAfter = Math.max(1, HostLoadStaleAfter.value()) * 1000L;
+        if (now - sample.updatedAt > staleAfter) {
+            // the host has stopped reporting; stop letting its last known figures speak for it
+            return HostLoad.UNKNOWN;
+        }
+        return sample.toHostLoad();
     }
 
     protected void clear() {
@@ -150,7 +210,7 @@ public String getConfigComponentName() {
 
     @Override
     public ConfigKey<?>[] getConfigKeys() {
-        return new ConfigKey<?>[] {HostLoadSampleInterval, HostLoadHalfLife};
+        return new ConfigKey<?>[] {HostLoadSampleInterval, HostLoadHalfLife, HostLoadStaleAfter};
     }
 
     /**
@@ -162,21 +222,28 @@ private static final class Sample {
         private final double memory;
         private final long updatedAt;
         private final long count;
+        private final HostStats reading;
 
-        private Sample(double cpu, double memory, long updatedAt) {
-            this(cpu, memory, updatedAt, 1);
+        private Sample(double cpu, double memory, long updatedAt, HostStats reading) {
+            this(cpu, memory, updatedAt, 1, reading);
         }
 
-        private Sample(double cpu, double memory, long updatedAt, long count) {
+        private Sample(double cpu, double memory, long updatedAt, long count, HostStats reading) {
             this.cpu = cpu;
             this.memory = memory;
             this.updatedAt = updatedAt;
             this.count = count;
+            this.reading = reading;
+        }
+
+        private boolean isSameReadingAs(HostStats stats) {
+            return reading == stats;
         }
 
-        private Sample fold(double newCpu, double newMemory, long now, int halfLifeSeconds) {
+        private Sample fold(double newCpu, double newMemory, long now, int halfLifeSeconds, HostStats reading) {
             double alpha = alpha(now - updatedAt, halfLifeSeconds);
-            return new Sample(cpu + alpha * (newCpu - cpu), memory + alpha * (newMemory - memory), now, count + 1);
+            return new Sample(cpu + alpha * (newCpu - cpu), memory + alpha * (newMemory - memory), now,
+                    count + 1, reading);
         }
 
         private static double alpha(long elapsedMillis, int halfLifeSeconds) {