Feature: Add support for DRS in a Cluster (#7723)

This pull request (PR) implements a Distributed Resource Scheduler (DRS) for a CloudStack cluster. The primary objective of this feature is to enable automatic resource optimization and workload balancing within the cluster by live migrating the VMs as per configuration.
Administrators can also execute DRS manually for a cluster, using the UI or the API.
Adds support for two algorithms - condensed & balanced. Algorithms are pluggable allowing ACS Administrators to have customized control over scheduling.

Implementation
There are three top level components:

    Scheduler
    A timer task which:

    Generate DRS plan for clusters
    Process DRS plan
    Remove old DRS plan records

    DRS Execution
    We go through each VM in the cluster and use the specified algorithm to check if DRS is required and to calculate cost, benefit & improvement of migrating that VM to another host in the cluster. On the basis of cost, benefit & improvement, the best migration is selected for the current iteration and the VM is migrated. The maximum number of iterations (live migrations) possible on the cluster is defined by drs.iterations which is defined as a percentage (as a value between 0 and 1) of total number of workloads.

    Algorithm
    Every algorithms implements two methods:
        needsDrs - to check if drs is required for cluster
        getMetrics - to calculate cost, benefit & improvement of a migrating a VM to another host.

Algorithms

    Condensed - Packs all the VMs on minimum number of hosts in the cluster.
    Balanced - Distributes the VMs evenly across hosts in the cluster.
    Algorithms use drs.level to decide the amount of imbalance to allow in the cluster.

APIs Added

listClusterDrsPlan

    id - ID of the DRS plan to list
    clusterid - to list plans for a cluster id

generateClusterDrsPlan

    id - cluster id
    iterations - The maximum number of iterations in a DRS job defined as a percentage (as a value between 0 and 1) of total number of workloads. Defaults to value of cluster's drs.iterations setting.

executeClusterDrsPlan

    id - ID of the cluster for which DRS plan is to be executed.
    migrateto - This parameter specifies the mapping between a vm and a host to migrate that VM. Format of this parameter: migrateto[vm-index].vm=<uuid>&migrateto[vm-index].host=<uuid>.

Config Keys Added

    ClusterDrsPlanExpireInterval
    Key drs.plan.expire.interval
    Scope Global
    Default Value 30 days
    Description The interval in days after which old DRS records will be cleaned up.

    ClusterDrsEnabled
    Key drs.automatic.enable
    Scope Cluster
    Default Value false
    Description Enable/disable automatic DRS on a cluster.

    ClusterDrsInterval
    Key drs.automatic.interval
    Scope Cluster
    Default Value 60 minutes
    Description The interval in minutes after which a periodic background thread will schedule DRS for a cluster.

    ClusterDrsIterations
    Key drs.max.migrations
    Scope Cluster
    Default Value 50
    Description Maximum number of live migrations in a DRS execution.

    ClusterDrsAlgorithm
    Key drs.algorithm
    Scope Cluster
    Default Value condensed
    Description DRS algorithm to execute on the cluster. This PR implements two algorithms - balanced & condensed.

    ClusterDrsLevel
    Key drs.imbalance
    Scope Cluster
    Default Value 0.5
    Description Percentage (as a value between 0.0 and 1.0) of imbalance allowed in the cluster. 1.0 means no imbalance
    is allowed and 0.0 means imbalance is allowed.

    ClusterDrsMetric
    Key drs.imbalance.metric
    Scope Cluster
    Default Value memory
    Description The cluster imbalance metric to use when checking the drs.imbalance.threshold. Possible values are memory and cpu.

Author: vishesh92

.github/workflows/ci.yml

@@ -59,7 +59,8 @@ jobs:
                   smoke/test_domain_network_offerings
                   smoke/test_domain_service_offerings
                   smoke/test_domain_vpc_offerings",
-                "smoke/test_dynamicroles
+                "smoke/test_cluster_drs
+                  smoke/test_dynamicroles
                   smoke/test_enable_account_settings_for_domain
                   smoke/test_enable_role_based_users_in_projects
                   smoke/test_events_resource

api/pom.xml

@@ -36,6 +36,11 @@
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-math3</artifactId>
+            <version>${cs.commons-math3.version}</version>
+        </dependency>
         <dependency>
             <groupId>javax.servlet</groupId>
             <artifactId>javax.servlet-api</artifactId>

api/src/main/java/com/cloud/event/EventTypes.java

@@ -678,6 +678,11 @@ public class EventTypes {
     //Usage related events
     public static final String EVENT_USAGE_REMOVE_USAGE_RECORDS = "USAGE.REMOVE.USAGE.RECORDS";
 
+    // DRS Events
+    public static final String EVENT_CLUSTER_DRS = "CLUSTER.DRS";
+    public static final String EVENT_CLUSTER_DRS_GENERATE = "CLUSTER.DRS.GENERATE";
+
+
     // Netscaler Service Package events
     public static final String EVENT_NETSCALER_SERVICEPACKAGE_ADD = "NETSCALER.SERVICEPACKAGE.ADD";
     public static final String EVENT_NETSCALER_SERVICEPACKAGE_DELETE = "NETSCALER.SERVICEPACKAGE.DELETE";

api/src/main/java/com/cloud/server/ManagementService.java

@@ -441,6 +441,8 @@ public interface ManagementService {
      */
     Ternary<Pair<List<? extends Host>, Integer>, List<? extends Host>, Map<Host, Boolean>> listHostsForMigrationOfVM(Long vmId, Long startIndex, Long pageSize, String keyword);
 
+    Ternary<Pair<List<? extends Host>, Integer>, List<? extends Host>, Map<Host, Boolean>> listHostsForMigrationOfVM(VirtualMachine vm, Long startIndex, Long pageSize, String keyword, List<VirtualMachine> vmList);
+
     /**
      * List storage pools for live migrating of a volume. The API returns list of all pools in the cluster to which the
      * volume can be migrated. Current pool is not included in the list. In case of vSphere datastore cluster storage pools,

api/src/main/java/com/cloud/vm/VirtualMachine.java

@@ -315,6 +315,9 @@ public boolean isUsedBySystem() {
     @Override
     Long getHostId();
 
+
+    void setHostId(Long hostId);
+
     /**
      * @return should HA be enabled for this machine?
      */

api/src/main/java/com/cloud/vm/VmDetailConstants.java

@@ -73,6 +73,8 @@ public interface VmDetailConstants {
 
     String CONFIG_DRIVE_LOCATION = "configDriveLocation";
 
+    String SKIP_DRS = "skipFromDRS";
+
     // VM import with nic, disk and custom params for custom compute offering
     String NIC = "nic";
     String NETWORK = "network";

api/src/main/java/org/apache/cloudstack/affinity/AffinityGroupProcessor.java

@@ -21,8 +21,11 @@
 import com.cloud.deploy.DeploymentPlanner.ExcludeList;
 import com.cloud.exception.AffinityConflictException;
 import com.cloud.utils.component.Adapter;
+import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
 
+import java.util.List;
+
 public interface AffinityGroupProcessor extends Adapter {
 
     /**
@@ -35,7 +38,12 @@ public interface AffinityGroupProcessor extends Adapter {
      *            deployment plan that tells you where it's being deployed to.
      * @param avoid
      *            avoid these data centers, pods, clusters, or hosts.
+     * @param vmList
+     *            list of virtual machines objects according to which the affinity group should be processed.
+     *            This can be used to process a theoretical state in some cases like generating DRS plans
      */
+    void process(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid, List<VirtualMachine> vmList) throws AffinityConflictException;
+
     void process(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid) throws AffinityConflictException;
 
     /**

api/src/main/java/org/apache/cloudstack/affinity/AffinityProcessorBase.java

@@ -21,14 +21,23 @@
 import com.cloud.deploy.DeploymentPlanner.ExcludeList;
 import com.cloud.exception.AffinityConflictException;
 import com.cloud.utils.component.AdapterBase;
+import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
 
+import java.util.Collections;
+import java.util.List;
+
 public class AffinityProcessorBase extends AdapterBase implements AffinityGroupProcessor {
 
     protected String _type;
 
     @Override
     public void process(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid) throws AffinityConflictException {
+        process(vm, plan, avoid, Collections.emptyList());
+    }
+
+    @Override
+    public void process(VirtualMachineProfile vm, DeploymentPlan plan, ExcludeList avoid, List<VirtualMachine> vmList) throws AffinityConflictException {
 
     }
 

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -292,6 +292,7 @@ public class ApiConstants {
     public static final String MIN_CPU_NUMBER = "mincpunumber";
     public static final String MIN_MEMORY = "minmemory";
     public static final String MIGRATION_TYPE = "migrationtype";
+    public static final String MIGRATIONS = "migrations";
     public static final String MEMORY = "memory";
     public static final String MODE = "mode";
     public static final String NAME = "name";
@@ -661,6 +662,7 @@ public class ApiConstants {
     public static final String SPECIFY_IP_RANGES = "specifyipranges";
     public static final String IS_SOURCE_NAT = "issourcenat";
     public static final String IS_STATIC_NAT = "isstaticnat";
+    public static final String ITERATIONS = "iterations";
     public static final String SORT_BY = "sortby";
     public static final String CHANGE_CIDR = "changecidr";
     public static final String PURPOSE = "purpose";

api/src/main/java/org/apache/cloudstack/api/command/admin/cluster/ExecuteClusterDrsPlanCmd.java

@@ -0,0 +1,147 @@
+/*
+ * 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 org.apache.cloudstack.api.command.admin.cluster;
+
+import com.cloud.event.EventTypes;
+import com.cloud.exception.InvalidParameterValueException;
+import com.cloud.host.Host;
+import com.cloud.user.Account;
+import com.cloud.utils.UuidUtils;
+import com.cloud.vm.VirtualMachine;
+import org.apache.cloudstack.api.APICommand;
+import org.apache.cloudstack.api.ApiCommandResourceType;
+import org.apache.cloudstack.api.ApiConstants;
+import org.apache.cloudstack.api.BaseAsyncCmd;
+import org.apache.cloudstack.api.Parameter;
+import org.apache.cloudstack.api.response.ClusterDrsPlanResponse;
+import org.apache.cloudstack.api.response.ClusterResponse;
+import org.apache.cloudstack.cluster.ClusterDrsService;
+import org.apache.commons.collections.MapUtils;
+
+import javax.inject.Inject;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+@APICommand(name = "executeClusterDrsPlan",
+            description = "Execute DRS for a cluster. If there is another plan in progress for the same cluster, " +
+                    "this command will fail.",
+            responseObject = ClusterDrsPlanResponse.class, since = "4.19.0", requestHasSensitiveInfo = false,
+            responseHasSensitiveInfo = false)
+public class ExecuteClusterDrsPlanCmd extends BaseAsyncCmd {
+
+    @Parameter(name = ApiConstants.ID, type = CommandType.UUID, entityType = ClusterResponse.class, required = true,
+               description = "ID of cluster")
+    private Long id;
+
+    @Parameter(
+            name = ApiConstants.MIGRATE_TO,
+            type = CommandType.MAP,
+            description = "Virtual Machine to destination host mapping. This parameter specifies the mapping between " +
+                    "a vm and a host to migrate that VM. clusterid is required if this parameter is set." +
+                    "Format of this parameter: migrateto[vm-index].vm=<uuid>&migrateto[vm-index].host=<uuid> " +
+                    "Where, [vm-index] indicates the index to identify the vm that you want to migrate, " +
+                    "vm=<uuid> indicates the UUID of the vm that you want to migrate, and " +
+                    "host=<uuid> indicates the UUID of the host where you want to migrate the vm. " +
+                    "Example: migrateto[0].vm=<71f43cd6-69b0-4d3b-9fbc-67f50963d60b>" +
+                    "&migrateto[0].host=<a382f181-3d2b-4413-b92d-b8931befa7e1>" +
+                    "&migrateto[1].vm=<88de0173-55c0-4c1c-a269-83d0279eeedf>" +
+                    "&migrateto[1].host=<95d6e97c-6766-4d67-9a30-c449c15011d1>" +
+                    "&migrateto[2].vm=<1b331390-59f2-4796-9993-bf11c6e76225>" +
+                    "&migrateto[2].host=<41fdb564-9d3b-447d-88ed-7628f7640cbc>")
+    private Map<String, String> migrateVmTo;
+
+    @Inject
+    private ClusterDrsService clusterDrsService;
+
+    public Map<VirtualMachine, Host> getVmToHostMap() {
+        Map<VirtualMachine, Host> vmToHostMap = new HashMap<>();
+        if (MapUtils.isNotEmpty(migrateVmTo)) {
+            Collection<?> allValues = migrateVmTo.values();
+            Iterator<?> iter = allValues.iterator();
+            while (iter.hasNext()) {
+                HashMap<String, String> vmToHost = (HashMap<String, String>) iter.next();
+
+                String vmId = vmToHost.get("vm");
+                String hostId = vmToHost.get("host");
+
+                VirtualMachine vm;
+                Host host;
+                if (UuidUtils.isUuid(vmId)) {
+                    vm = _entityMgr.findByUuid(VirtualMachine.class, vmId);
+                } else {
+                    vm = _entityMgr.findById(VirtualMachine.class, Long.parseLong(vmId));
+                }
+
+                if (UuidUtils.isUuid(hostId)) {
+                    host = _entityMgr.findByUuid(Host.class, hostId);
+                } else {
+                    host = _entityMgr.findById(Host.class, Long.parseLong(hostId));
+                }
+
+                if (vm == null || host == null) {
+                    throw new InvalidParameterValueException(
+                            String.format("Unable to find the vm/host for vmId=%s, destHostId=%s", vmId, hostId));
+                }
+
+                vmToHostMap.put(vm, host);
+            }
+        }
+        return vmToHostMap;
+    }
+
+    @Override
+    public void execute() {
+        ClusterDrsPlanResponse response = clusterDrsService.executeDrsPlan(this);
+        response.setResponseName(getCommandName());
+        this.setResponseObject(response);
+    }
+
+    @Override
+    public long getEntityOwnerId() {
+        return Account.ACCOUNT_ID_SYSTEM;
+    }
+
+    @Override
+    public Long getApiResourceId() {
+        return getId();
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    @Override
+    public ApiCommandResourceType getApiResourceType() {
+        return ApiCommandResourceType.Cluster;
+    }
+
+
+    @Override
+    public String getEventType() {
+        return EventTypes.EVENT_CLUSTER_DRS;
+    }
+
+    @Override
+    public String getEventDescription() {
+        return String.format("Executing DRS plan for cluster: %d", getId());
+    }
+}