cluster-drs: trigger DRS on VM power-state events and add a combined cpu+memory metric

api/src/main/java/org/apache/cloudstack/cluster/ClusterDrsAlgorithm.java

@@ -96,9 +96,19 @@ Ternary<Double, Double, Double> getMetrics(Cluster cluster, VirtualMachine vm, S
      * @return the cluster imbalance after migration
      */
     default Double getImbalancePostMigration(VirtualMachine vm,
-            Host destHost, Long clusterId, long vmMetric, double[] baseMetricsArray,
+            Host destHost, Long clusterId, ServiceOffering serviceOffering, double[] baseMetricsArray,
             Map<Long, Integer> hostIdToIndexMap, Map<Long, Ternary<Long, Long, Long>> hostCpuMap,
-            Map<Long, Ternary<Long, Long, Long>> hostMemoryMap) {
+            Map<Long, Ternary<Long, Long, Long>> hostMemoryMap) throws ConfigurationException {
+        // Metric "both": return the worse of the cpu and memory post-migration imbalance. The
+        // baseMetricsArray fast path is single-metric, so evaluate both maps directly instead.
+        if ("both".equals(getClusterDrsMetric(clusterId))) {
+            long vmCpuMetric = (long) serviceOffering.getCpu() * serviceOffering.getSpeed();
+            long vmMemMetric = serviceOffering.getRamSize() * 1024L * 1024L;
+            return Math.max(
+                    imbalancePostMigrationForMap(vm, destHost, clusterId, vmCpuMetric, hostCpuMap),
+                    imbalancePostMigrationForMap(vm, destHost, clusterId, vmMemMetric, hostMemoryMap));
+        }
+        long vmMetric = getVmMetric(serviceOffering, clusterId);
         // Create a copy of the base array and adjust only the two affected hosts
         double[] adjustedMetrics = new double[baseMetricsArray.length];
         System.arraycopy(baseMetricsArray, 0, adjustedMetrics, 0, baseMetricsArray.length);
@@ -129,6 +139,23 @@ default Double getImbalancePostMigration(VirtualMachine vm,
         return calculateImbalance(adjustedMetrics);
     }
 
+    /**
+     * Cluster imbalance for a single resource map after hypothetically migrating vm to destHost.
+     */
+    private Double imbalancePostMigrationForMap(VirtualMachine vm, Host destHost, Long clusterId,
+            long vmMetric, Map<Long, Ternary<Long, Long, Long>> metricMap) {
+        long destHostId = destHost.getId();
+        long vmHostId = vm.getHostId();
+        List<Double> list = new ArrayList<>();
+        for (Map.Entry<Long, Ternary<Long, Long, Long>> entry : metricMap.entrySet()) {
+            Double value = getMetricValuePostMigration(clusterId, entry.getValue(), vmMetric, entry.getKey(), destHostId, vmHostId);
+            if (value != null) {
+                list.add(value);
+            }
+        }
+        return getImbalance(list);
+    }
+
     /**
      * Calculate imbalance from an array of metric values.
      * Imbalance is defined as standard deviation divided by mean.
@@ -272,6 +299,10 @@ static Double getClusterImbalance(Long clusterId, List<Ternary<Long, Long, Long>
             case "memory":
                 list = getMetricList(clusterId, memoryList, skipThreshold);
                 break;
+            case "both":
+                return Math.max(
+                        getImbalance(getMetricList(clusterId, cpuList, skipThreshold)),
+                        getImbalance(getMetricList(clusterId, memoryList, skipThreshold)));
             default:
                 throw new ConfigurationException(
                         String.format("Invalid metric: %s for cluster: %d", metric, clusterId));

api/src/main/java/org/apache/cloudstack/cluster/ClusterDrsService.java

@@ -45,6 +45,20 @@ public interface ClusterDrsService extends Manager, Configurable, Scheduler {
             "The interval in minutes after which a periodic background thread will schedule DRS for a cluster.", true,
             ConfigKey.Scope.Cluster, null, "Interval for Automatic DRS ", null, null, null);
 
+    ConfigKey<Boolean> ClusterDrsEventDrivenEnabled = new ConfigKey<>(Boolean.class, "drs.event.driven.enable",
+            ConfigKey.CATEGORY_ADVANCED, "false",
+            "In addition to the periodic drs.automatic.interval timer, trigger DRS for a cluster on VM " +
+                    "power-state events (deploy, start, stop, migrate) so it reacts to imbalance in near-real-time " +
+                    "rather than waiting a whole interval. Requires drs.automatic.enable; rate-limited per cluster " +
+                    "by drs.event.driven.interval.", true,
+            ConfigKey.Scope.Cluster, null, "Enable event-driven DRS", null, null, null);
+
+    ConfigKey<Integer> ClusterDrsEventDrivenInterval = new ConfigKey<>(Integer.class, "drs.event.driven.interval",
+            ConfigKey.CATEGORY_ADVANCED, "5",
+            "Minimum minutes between event-triggered DRS runs for a cluster (debounce). " +
+                    "Only applies when drs.event.driven.enable is true.", true,
+            ConfigKey.Scope.Cluster, null, "Event-driven DRS min interval", null, null, null);
+
     ConfigKey<Integer> ClusterDrsMaxMigrations = new ConfigKey<>(Integer.class, "drs.max.migrations",
             ConfigKey.CATEGORY_ADVANCED, "50",
             "Maximum number of live migrations in a DRS execution.",
@@ -62,9 +76,10 @@ public interface ClusterDrsService extends Manager, Configurable, Scheduler {
 
     ConfigKey<String> ClusterDrsMetric = new ConfigKey<>(String.class, "drs.metric", ConfigKey.CATEGORY_ADVANCED,
             "memory",
-            "The allocated resource metric used to measure imbalance in a cluster. Possible values are memory, cpu.",
+            "The allocated resource metric used to measure imbalance in a cluster. Possible values are memory, cpu, both. " +
+                    "'both' balances on the worse of the cpu and memory imbalance so neither resource is left contended.",
             true, ConfigKey.Scope.Cluster, null, "DRS metric", null, null, null, ConfigKey.Kind.Select,
-            "memory,cpu");
+            "memory,cpu,both");
 
     ConfigKey<String> ClusterDrsMetricType = new ConfigKey<>(String.class, "drs.metric.type", ConfigKey.CATEGORY_ADVANCED,
             "used",

plugins/drs/cluster/balanced/src/main/java/org/apache/cloudstack/cluster/Balanced.java

@@ -82,7 +82,7 @@ public Ternary<Double, Double, Double> getMetrics(Cluster cluster, VirtualMachin
 
         // Use optimized post-imbalance calculation that adjusts only affected hosts
         Double postImbalance = getImbalancePostMigration(vm, destHost,
-                cluster.getId(), ClusterDrsAlgorithm.getVmMetric(serviceOffering, cluster.getId()),
+                cluster.getId(), serviceOffering,
                 baseMetricsArray, hostIdToIndexMap, hostCpuMap, hostMemoryMap);
 
         logger.trace("Cluster {} pre-imbalance: {} post-imbalance: {} Algorithm: {} VM: {} srcHost ID: {} destHost: {}",

plugins/drs/cluster/condensed/src/main/java/org/apache/cloudstack/cluster/Condensed.java

@@ -85,7 +85,7 @@ public Ternary<Double, Double, Double> getMetrics(Cluster cluster, VirtualMachin
 
         // Use optimized post-imbalance calculation that adjusts only affected hosts
         Double postImbalance = getImbalancePostMigration(vm, destHost,
-                cluster.getId(), ClusterDrsAlgorithm.getVmMetric(serviceOffering, cluster.getId()),
+                cluster.getId(), serviceOffering,
                 baseMetricsArray, hostIdToIndexMap, hostCpuMap, hostMemoryMap);
 
         logger.trace("Cluster {} pre-imbalance: {} post-imbalance: {} Algorithm: {} VM: {} srcHost ID: {} destHost: {}",

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

@@ -53,6 +53,14 @@
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.VMInstanceDetailVO;
 import com.cloud.vm.VMInstanceVO;
+import com.cloud.vm.VirtualMachineManager;
+import com.cloud.utils.concurrency.NamedThreadFactory;
+import org.apache.cloudstack.framework.messagebus.MessageBus;
+import org.apache.cloudstack.framework.messagebus.MessageDispatcher;
+import org.apache.cloudstack.framework.messagebus.MessageHandler;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
 import com.cloud.vm.VirtualMachineProfileImpl;
@@ -148,6 +156,13 @@ public class ClusterDrsServiceImpl extends ManagerBase implements ClusterDrsServ
 
     Map<String, ClusterDrsAlgorithm> drsAlgorithmMap = new HashMap<>();
 
+    @Inject
+    MessageBus messageBus;
+    // Epoch-ms of the last event-triggered DRS run, per cluster; drives the cooldown.
+    private final Map<Long, Long> lastEventDrsTriggerByCluster = new ConcurrentHashMap<>();
+    // Runs event-triggered DRS off the message-bus thread.
+    private ExecutorService eventDrsExecutor;
+
     public AsyncJobDispatcher getAsyncJobDispatcher() {
         return asyncJobDispatcher;
     }
@@ -179,6 +194,11 @@ protected void runInContext() {
         };
         Timer vmSchedulerTimer = new Timer("VMSchedulerPollTask");
         vmSchedulerTimer.schedule(schedulerPollTask, 5000L, 60 * 1000L);
+
+        // Subscribe to VM power-state events for event-driven DRS (gated per cluster by drs.event.driven.enable).
+        eventDrsExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("Event-Driven-DRS"));
+        messageBus.subscribe(VirtualMachineManager.Topics.VM_POWER_STATE, MessageDispatcher.getDispatcher(this));
+
         return true;
     }
 
@@ -278,48 +298,127 @@ void generateDrsPlanForAllClusters() {
         List<ClusterVO> clusterList = clusterDao.listAll();
 
         for (ClusterVO cluster : clusterList) {
-            if (cluster.getAllocationState() == Disabled || ClusterDrsEnabled.valueIn(
-                    cluster.getId()).equals(Boolean.FALSE)) {
-                continue;
-            }
+            generateDrsPlanForCluster(cluster, ClusterDrsInterval.valueIn(cluster.getId()));
+        }
+    }
 
-            ClusterDrsPlanVO lastPlan = drsPlanDao.listLatestPlanForClusterId(cluster.getId());
-
-            // If the last plan is ready or in progress or was executed within the last interval, skip this cluster.
-            // This is to avoid generating plans for clusters which are already being processed and to avoid
-            // generating plans for clusters which have been processed recently.This doesn't consider the type
-            // (manual or automated) of the last plan.
-            if (lastPlan != null && (lastPlan.getStatus() == ClusterDrsPlan.Status.READY ||
-                    lastPlan.getStatus() == ClusterDrsPlan.Status.IN_PROGRESS ||
-                    (lastPlan.getStatus() == ClusterDrsPlan.Status.COMPLETED &&
-                            lastPlan.getCreated().compareTo(DateUtils.addMinutes(new Date(), -1 * ClusterDrsInterval.valueIn(cluster.getId()))) > 0)
-            )) {
-                continue;
+    /**
+     * Generates a DRS plan for a single cluster, skipping if DRS is disabled, a plan is already
+     * pending/running, or one completed within {@code debounceMinutes}.
+     */
+    void generateDrsPlanForCluster(ClusterVO cluster, int debounceMinutes) {
+        if (cluster.getAllocationState() == Disabled || ClusterDrsEnabled.valueIn(cluster.getId()).equals(Boolean.FALSE)) {
+            return;
+        }
+
+        ClusterDrsPlanVO lastPlan = drsPlanDao.listLatestPlanForClusterId(cluster.getId());
+
+        // Skip if the last plan is ready, in progress, or completed within the debounce window.
+        if (lastPlan != null && (lastPlan.getStatus() == ClusterDrsPlan.Status.READY ||
+                lastPlan.getStatus() == ClusterDrsPlan.Status.IN_PROGRESS ||
+                (lastPlan.getStatus() == ClusterDrsPlan.Status.COMPLETED &&
+                        lastPlan.getCreated().compareTo(DateUtils.addMinutes(new Date(), -1 * debounceMinutes)) > 0)
+        )) {
+            return;
+        }
+
+        long eventId = ActionEventUtils.onStartedActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM,
+                EventTypes.EVENT_CLUSTER_DRS,
+                String.format("Generating DRS plan for cluster %s", cluster.getUuid()), cluster.getId(),
+                ApiCommandResourceType.Cluster.toString(), true, 0);
+        GlobalLock clusterLock = GlobalLock.getInternLock(String.format(CLUSTER_LOCK_STR, cluster.getId()));
+        try {
+            if (clusterLock.lock(30)) {
+                try {
+                    List<Ternary<VirtualMachine, Host, Host>> plan = getDrsPlan(cluster,
+                            ClusterDrsMaxMigrations.valueIn(cluster.getId()));
+                    savePlan(cluster.getId(), plan, eventId, ClusterDrsPlan.Type.AUTOMATED,
+                            ClusterDrsPlan.Status.READY);
+                    logger.info("Generated DRS plan for cluster {}", cluster);
+                } catch (Exception e) {
+                    logger.error("Unable to generate DRS plans for cluster {}", cluster, e);
+                } finally {
+                    clusterLock.unlock();
+                }
             }
+        } finally {
+            clusterLock.releaseRef();
+        }
+    }
 
-            long eventId = ActionEventUtils.onStartedActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM,
-                    EventTypes.EVENT_CLUSTER_DRS,
-                    String.format("Generating DRS plan for cluster %s", cluster.getUuid()), cluster.getId(),
-                    ApiCommandResourceType.Cluster.toString(), true, 0);
-            GlobalLock clusterLock = GlobalLock.getInternLock(String.format(CLUSTER_LOCK_STR, cluster.getId()));
+    /**
+     * Message-bus handler for VM power-state events; triggers event-driven DRS for the VM's cluster.
+     */
+    @MessageHandler(topic = VirtualMachineManager.Topics.VM_POWER_STATE)
+    protected void handleVmPowerStateEvent(String subject, String senderAddress, Object args) {
+        if (!(args instanceof Long)) {
+            return;
+        }
+        try {
+            triggerEventDrivenDrsForVm((Long) args);
+        } catch (Exception e) {
+            logger.debug("Event-driven DRS: error handling VM power-state event for {}", args, e);
+        }
+    }
+
+    /**
+     * Resolves the VM's cluster and, subject to the enable flag and cooldown, schedules DRS plan generation.
+     */
+    void triggerEventDrivenDrsForVm(Long vmId) {
+        if (vmId == null) {
+            return;
+        }
+        VMInstanceVO vm = vmInstanceDao.findById(vmId);
+        if (vm == null || vm.getHostId() == null) {
+            return;
+        }
+        HostVO host = hostDao.findById(vm.getHostId());
+        if (host == null || host.getClusterId() == null) {
+            return;
+        }
+        Long clusterId = host.getClusterId();
+        if (!shouldTriggerEventDrivenDrs(clusterId)) {
+            return;
+        }
+        final ClusterVO cluster = clusterDao.findById(clusterId);
+        if (cluster == null) {
+            return;
+        }
+        final int debounceMinutes = ClusterDrsEventDrivenInterval.valueIn(clusterId);
+        logger.debug("Event-driven DRS: scheduling plan generation for cluster {} (triggered by VM {})", clusterId, vmId);
+        submitEventDrivenDrs(cluster, debounceMinutes);
+    }
+
+    /**
+     * Runs generateDrsPlanForCluster off the message-bus thread so event publishers are not blocked.
+     */
+    protected void submitEventDrivenDrs(final ClusterVO cluster, final int debounceMinutes) {
+        eventDrsExecutor.submit(() -> {
             try {
-                if (clusterLock.lock(30)) {
-                    try {
-                        List<Ternary<VirtualMachine, Host, Host>> plan = getDrsPlan(cluster,
-                                ClusterDrsMaxMigrations.valueIn(cluster.getId()));
-                        savePlan(cluster.getId(), plan, eventId, ClusterDrsPlan.Type.AUTOMATED,
-                                ClusterDrsPlan.Status.READY);
-                        logger.info("Generated DRS plan for cluster {}", cluster);
-                    } catch (Exception e) {
-                        logger.error("Unable to generate DRS plans for cluster {}", cluster, e);
-                    } finally {
-                        clusterLock.unlock();
-                    }
-                }
-            } finally {
-                clusterLock.releaseRef();
+                generateDrsPlanForCluster(cluster, debounceMinutes);
+            } catch (Exception e) {
+                logger.warn("Event-driven DRS: plan generation failed for cluster {}", cluster, e);
             }
+        });
+    }
+
+    /**
+     * Returns true if automatic and event-driven DRS are enabled and the per-cluster cooldown has
+     * elapsed, recording the trigger time when it does.
+     */
+    boolean shouldTriggerEventDrivenDrs(Long clusterId) {
+        if (ClusterDrsEnabled.valueIn(clusterId).equals(Boolean.FALSE)
+                || ClusterDrsEventDrivenEnabled.valueIn(clusterId).equals(Boolean.FALSE)) {
+            return false;
         }
+        long now = System.currentTimeMillis();
+        long cooldownMs = ClusterDrsEventDrivenInterval.valueIn(clusterId) * 60L * 1000L;
+        Long last = lastEventDrsTriggerByCluster.get(clusterId);
+        if (last != null && (now - last) < cooldownMs) {
+            return false;
+        }
+        lastEventDrsTriggerByCluster.put(clusterId, now);
+        return true;
     }
 
     /**
@@ -855,7 +954,7 @@ public String getConfigComponentName() {
     public ConfigKey<?>[] getConfigKeys() {
         return new ConfigKey<?>[]{ClusterDrsPlanExpireInterval, ClusterDrsEnabled, ClusterDrsInterval, ClusterDrsMaxMigrations,
                 ClusterDrsAlgorithm, ClusterDrsImbalanceThreshold, ClusterDrsMetric, ClusterDrsMetricType, ClusterDrsMetricUseRatio,
-                ClusterDrsImbalanceSkipThreshold};
+                ClusterDrsImbalanceSkipThreshold, ClusterDrsEventDrivenEnabled, ClusterDrsEventDrivenInterval};
     }
 
     @Override