Skip to content

Commit 216282d

Browse files
rootnomadrootnomad
committed
drs: trigger DRS on VM power-state events, not only the timer
Adds drs.event.driven.enable (default false) and drs.event.driven.interval (default 5 min). When enabled for a cluster, ClusterDrsServiceImpl subscribes to the vm.powerstate message bus, resolves the affected cluster on each event and generates a plan for it within minutes (rate- limited per cluster), off the message-bus thread. The existing poll then executes the plan. Requires drs.automatic.enable.
1 parent 7772118 commit 216282d

3 files changed

Lines changed: 261 additions & 37 deletions

File tree

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,20 @@ public interface ClusterDrsService extends Manager, Configurable, Scheduler {
4545
"The interval in minutes after which a periodic background thread will schedule DRS for a cluster.", true,
4646
ConfigKey.Scope.Cluster, null, "Interval for Automatic DRS ", null, null, null);
4747

48+
ConfigKey<Boolean> ClusterDrsEventDrivenEnabled = new ConfigKey<>(Boolean.class, "drs.event.driven.enable",
49+
ConfigKey.CATEGORY_ADVANCED, "false",
50+
"In addition to the periodic drs.automatic.interval timer, trigger DRS for a cluster on VM " +
51+
"power-state events (deploy, start, stop, migrate) so it reacts to imbalance in near-real-time " +
52+
"rather than waiting a whole interval. Requires drs.automatic.enable; rate-limited per cluster " +
53+
"by drs.event.driven.interval.", true,
54+
ConfigKey.Scope.Cluster, null, "Enable event-driven DRS", null, null, null);
55+
56+
ConfigKey<Integer> ClusterDrsEventDrivenInterval = new ConfigKey<>(Integer.class, "drs.event.driven.interval",
57+
ConfigKey.CATEGORY_ADVANCED, "5",
58+
"Minimum minutes between event-triggered DRS runs for a cluster (debounce). " +
59+
"Only applies when drs.event.driven.enable is true.", true,
60+
ConfigKey.Scope.Cluster, null, "Event-driven DRS min interval", null, null, null);
61+
4862
ConfigKey<Integer> ClusterDrsMaxMigrations = new ConfigKey<>(Integer.class, "drs.max.migrations",
4963
ConfigKey.CATEGORY_ADVANCED, "50",
5064
"Maximum number of live migrations in a DRS execution.",

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

Lines changed: 136 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,14 @@
5353
import com.cloud.utils.exception.CloudRuntimeException;
5454
import com.cloud.vm.VMInstanceDetailVO;
5555
import com.cloud.vm.VMInstanceVO;
56+
import com.cloud.vm.VirtualMachineManager;
57+
import com.cloud.utils.concurrency.NamedThreadFactory;
58+
import org.apache.cloudstack.framework.messagebus.MessageBus;
59+
import org.apache.cloudstack.framework.messagebus.MessageDispatcher;
60+
import org.apache.cloudstack.framework.messagebus.MessageHandler;
61+
import java.util.concurrent.ConcurrentHashMap;
62+
import java.util.concurrent.ExecutorService;
63+
import java.util.concurrent.Executors;
5664
import com.cloud.vm.VirtualMachine;
5765
import com.cloud.vm.VirtualMachineProfile;
5866
import com.cloud.vm.VirtualMachineProfileImpl;
@@ -148,6 +156,13 @@ public class ClusterDrsServiceImpl extends ManagerBase implements ClusterDrsServ
148156

149157
Map<String, ClusterDrsAlgorithm> drsAlgorithmMap = new HashMap<>();
150158

159+
@Inject
160+
MessageBus messageBus;
161+
// Epoch-ms of the last event-triggered DRS run, per cluster; drives the cooldown.
162+
private final Map<Long, Long> lastEventDrsTriggerByCluster = new ConcurrentHashMap<>();
163+
// Runs event-triggered DRS off the message-bus thread.
164+
private ExecutorService eventDrsExecutor;
165+
151166
public AsyncJobDispatcher getAsyncJobDispatcher() {
152167
return asyncJobDispatcher;
153168
}
@@ -179,6 +194,11 @@ protected void runInContext() {
179194
};
180195
Timer vmSchedulerTimer = new Timer("VMSchedulerPollTask");
181196
vmSchedulerTimer.schedule(schedulerPollTask, 5000L, 60 * 1000L);
197+
198+
// Subscribe to VM power-state events for event-driven DRS (gated per cluster by drs.event.driven.enable).
199+
eventDrsExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("Event-Driven-DRS"));
200+
messageBus.subscribe(VirtualMachineManager.Topics.VM_POWER_STATE, MessageDispatcher.getDispatcher(this));
201+
182202
return true;
183203
}
184204

@@ -278,48 +298,127 @@ void generateDrsPlanForAllClusters() {
278298
List<ClusterVO> clusterList = clusterDao.listAll();
279299

280300
for (ClusterVO cluster : clusterList) {
281-
if (cluster.getAllocationState() == Disabled || ClusterDrsEnabled.valueIn(
282-
cluster.getId()).equals(Boolean.FALSE)) {
283-
continue;
284-
}
301+
generateDrsPlanForCluster(cluster, ClusterDrsInterval.valueIn(cluster.getId()));
302+
}
303+
}
285304

286-
ClusterDrsPlanVO lastPlan = drsPlanDao.listLatestPlanForClusterId(cluster.getId());
287-
288-
// If the last plan is ready or in progress or was executed within the last interval, skip this cluster.
289-
// This is to avoid generating plans for clusters which are already being processed and to avoid
290-
// generating plans for clusters which have been processed recently.This doesn't consider the type
291-
// (manual or automated) of the last plan.
292-
if (lastPlan != null && (lastPlan.getStatus() == ClusterDrsPlan.Status.READY ||
293-
lastPlan.getStatus() == ClusterDrsPlan.Status.IN_PROGRESS ||
294-
(lastPlan.getStatus() == ClusterDrsPlan.Status.COMPLETED &&
295-
lastPlan.getCreated().compareTo(DateUtils.addMinutes(new Date(), -1 * ClusterDrsInterval.valueIn(cluster.getId()))) > 0)
296-
)) {
297-
continue;
305+
/**
306+
* Generates a DRS plan for a single cluster, skipping if DRS is disabled, a plan is already
307+
* pending/running, or one completed within {@code debounceMinutes}.
308+
*/
309+
void generateDrsPlanForCluster(ClusterVO cluster, int debounceMinutes) {
310+
if (cluster.getAllocationState() == Disabled || ClusterDrsEnabled.valueIn(cluster.getId()).equals(Boolean.FALSE)) {
311+
return;
312+
}
313+
314+
ClusterDrsPlanVO lastPlan = drsPlanDao.listLatestPlanForClusterId(cluster.getId());
315+
316+
// Skip if the last plan is ready, in progress, or completed within the debounce window.
317+
if (lastPlan != null && (lastPlan.getStatus() == ClusterDrsPlan.Status.READY ||
318+
lastPlan.getStatus() == ClusterDrsPlan.Status.IN_PROGRESS ||
319+
(lastPlan.getStatus() == ClusterDrsPlan.Status.COMPLETED &&
320+
lastPlan.getCreated().compareTo(DateUtils.addMinutes(new Date(), -1 * debounceMinutes)) > 0)
321+
)) {
322+
return;
323+
}
324+
325+
long eventId = ActionEventUtils.onStartedActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM,
326+
EventTypes.EVENT_CLUSTER_DRS,
327+
String.format("Generating DRS plan for cluster %s", cluster.getUuid()), cluster.getId(),
328+
ApiCommandResourceType.Cluster.toString(), true, 0);
329+
GlobalLock clusterLock = GlobalLock.getInternLock(String.format(CLUSTER_LOCK_STR, cluster.getId()));
330+
try {
331+
if (clusterLock.lock(30)) {
332+
try {
333+
List<Ternary<VirtualMachine, Host, Host>> plan = getDrsPlan(cluster,
334+
ClusterDrsMaxMigrations.valueIn(cluster.getId()));
335+
savePlan(cluster.getId(), plan, eventId, ClusterDrsPlan.Type.AUTOMATED,
336+
ClusterDrsPlan.Status.READY);
337+
logger.info("Generated DRS plan for cluster {}", cluster);
338+
} catch (Exception e) {
339+
logger.error("Unable to generate DRS plans for cluster {}", cluster, e);
340+
} finally {
341+
clusterLock.unlock();
342+
}
298343
}
344+
} finally {
345+
clusterLock.releaseRef();
346+
}
347+
}
299348

300-
long eventId = ActionEventUtils.onStartedActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM,
301-
EventTypes.EVENT_CLUSTER_DRS,
302-
String.format("Generating DRS plan for cluster %s", cluster.getUuid()), cluster.getId(),
303-
ApiCommandResourceType.Cluster.toString(), true, 0);
304-
GlobalLock clusterLock = GlobalLock.getInternLock(String.format(CLUSTER_LOCK_STR, cluster.getId()));
349+
/**
350+
* Message-bus handler for VM power-state events; triggers event-driven DRS for the VM's cluster.
351+
*/
352+
@MessageHandler(topic = VirtualMachineManager.Topics.VM_POWER_STATE)
353+
protected void handleVmPowerStateEvent(String subject, String senderAddress, Object args) {
354+
if (!(args instanceof Long)) {
355+
return;
356+
}
357+
try {
358+
triggerEventDrivenDrsForVm((Long) args);
359+
} catch (Exception e) {
360+
logger.debug("Event-driven DRS: error handling VM power-state event for {}", args, e);
361+
}
362+
}
363+
364+
/**
365+
* Resolves the VM's cluster and, subject to the enable flag and cooldown, schedules DRS plan generation.
366+
*/
367+
void triggerEventDrivenDrsForVm(Long vmId) {
368+
if (vmId == null) {
369+
return;
370+
}
371+
VMInstanceVO vm = vmInstanceDao.findById(vmId);
372+
if (vm == null || vm.getHostId() == null) {
373+
return;
374+
}
375+
HostVO host = hostDao.findById(vm.getHostId());
376+
if (host == null || host.getClusterId() == null) {
377+
return;
378+
}
379+
Long clusterId = host.getClusterId();
380+
if (!shouldTriggerEventDrivenDrs(clusterId)) {
381+
return;
382+
}
383+
final ClusterVO cluster = clusterDao.findById(clusterId);
384+
if (cluster == null) {
385+
return;
386+
}
387+
final int debounceMinutes = ClusterDrsEventDrivenInterval.valueIn(clusterId);
388+
logger.debug("Event-driven DRS: scheduling plan generation for cluster {} (triggered by VM {})", clusterId, vmId);
389+
submitEventDrivenDrs(cluster, debounceMinutes);
390+
}
391+
392+
/**
393+
* Runs generateDrsPlanForCluster off the message-bus thread so event publishers are not blocked.
394+
*/
395+
protected void submitEventDrivenDrs(final ClusterVO cluster, final int debounceMinutes) {
396+
eventDrsExecutor.submit(() -> {
305397
try {
306-
if (clusterLock.lock(30)) {
307-
try {
308-
List<Ternary<VirtualMachine, Host, Host>> plan = getDrsPlan(cluster,
309-
ClusterDrsMaxMigrations.valueIn(cluster.getId()));
310-
savePlan(cluster.getId(), plan, eventId, ClusterDrsPlan.Type.AUTOMATED,
311-
ClusterDrsPlan.Status.READY);
312-
logger.info("Generated DRS plan for cluster {}", cluster);
313-
} catch (Exception e) {
314-
logger.error("Unable to generate DRS plans for cluster {}", cluster, e);
315-
} finally {
316-
clusterLock.unlock();
317-
}
318-
}
319-
} finally {
320-
clusterLock.releaseRef();
398+
generateDrsPlanForCluster(cluster, debounceMinutes);
399+
} catch (Exception e) {
400+
logger.warn("Event-driven DRS: plan generation failed for cluster {}", cluster, e);
321401
}
402+
});
403+
}
404+
405+
/**
406+
* Returns true if automatic and event-driven DRS are enabled and the per-cluster cooldown has
407+
* elapsed, recording the trigger time when it does.
408+
*/
409+
boolean shouldTriggerEventDrivenDrs(Long clusterId) {
410+
if (ClusterDrsEnabled.valueIn(clusterId).equals(Boolean.FALSE)
411+
|| ClusterDrsEventDrivenEnabled.valueIn(clusterId).equals(Boolean.FALSE)) {
412+
return false;
322413
}
414+
long now = System.currentTimeMillis();
415+
long cooldownMs = ClusterDrsEventDrivenInterval.valueIn(clusterId) * 60L * 1000L;
416+
Long last = lastEventDrsTriggerByCluster.get(clusterId);
417+
if (last != null && (now - last) < cooldownMs) {
418+
return false;
419+
}
420+
lastEventDrsTriggerByCluster.put(clusterId, now);
421+
return true;
323422
}
324423

325424
/**
@@ -855,7 +954,7 @@ public String getConfigComponentName() {
855954
public ConfigKey<?>[] getConfigKeys() {
856955
return new ConfigKey<?>[]{ClusterDrsPlanExpireInterval, ClusterDrsEnabled, ClusterDrsInterval, ClusterDrsMaxMigrations,
857956
ClusterDrsAlgorithm, ClusterDrsImbalanceThreshold, ClusterDrsMetric, ClusterDrsMetricType, ClusterDrsMetricUseRatio,
858-
ClusterDrsImbalanceSkipThreshold};
957+
ClusterDrsImbalanceSkipThreshold, ClusterDrsEventDrivenEnabled, ClusterDrsEventDrivenInterval};
859958
}
860959

861960
@Override

‎server/src/test/java/org/apache/cloudstack/cluster/ClusterDrsServiceImplTest.java‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@
7878
import static org.junit.Assert.assertEquals;
7979
import static org.junit.Assert.assertFalse;
8080
import static org.junit.Assert.assertNull;
81+
import static org.junit.Assert.assertTrue;
8182

8283
@RunWith(MockitoJUnitRunner.class)
8384
public class ClusterDrsServiceImplTest {
@@ -951,4 +952,114 @@ public void testProcessPlans() {
951952

952953
Mockito.verify(clusterDrsService, Mockito.times(2)).executeDrsPlan(Mockito.any(ClusterDrsPlanVO.class));
953954
}
955+
956+
// ---- event-driven DRS ----
957+
// The ConfigKeys are shared interface constants, so each test that overrides a default restores it
958+
// in a finally block to avoid leaking into other tests.
959+
960+
private static String getConfigDefault(ConfigKey<?> key) throws Exception {
961+
Field f = ConfigKey.class.getDeclaredField("_defaultValue");
962+
f.setAccessible(true);
963+
return (String) f.get(key);
964+
}
965+
966+
private static void setConfigDefault(ConfigKey<?> key, String value) throws Exception {
967+
Field f = ConfigKey.class.getDeclaredField("_defaultValue");
968+
f.setAccessible(true);
969+
f.set(key, value);
970+
}
971+
972+
@Test
973+
public void testShouldTriggerEventDrivenDrsDisabledByDefault() throws Exception {
974+
// Automatic DRS enabled but event-driven off -> must not trigger.
975+
String origDrs = getConfigDefault(clusterDrsService.ClusterDrsEnabled);
976+
String origEvt = getConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled);
977+
try {
978+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, "true");
979+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, "false");
980+
assertFalse(clusterDrsService.shouldTriggerEventDrivenDrs(1L));
981+
} finally {
982+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, origDrs);
983+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, origEvt);
984+
}
985+
}
986+
987+
@Test
988+
public void testShouldTriggerEventDrivenDrsRequiresAutomaticDrs() throws Exception {
989+
// Event-driven on but automatic DRS off -> must not trigger (event-driven depends on drs.automatic.enable).
990+
String origDrs = getConfigDefault(clusterDrsService.ClusterDrsEnabled);
991+
String origEvt = getConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled);
992+
try {
993+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, "false");
994+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, "true");
995+
assertFalse(clusterDrsService.shouldTriggerEventDrivenDrs(1L));
996+
} finally {
997+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, origDrs);
998+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, origEvt);
999+
}
1000+
}
1001+
1002+
@Test
1003+
public void testShouldTriggerEventDrivenDrsEnabledThenDebounced() throws Exception {
1004+
String origDrs = getConfigDefault(clusterDrsService.ClusterDrsEnabled);
1005+
String origEvt = getConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled);
1006+
String origInt = getConfigDefault(clusterDrsService.ClusterDrsEventDrivenInterval);
1007+
try {
1008+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, "true");
1009+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, "true");
1010+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenInterval, "5");
1011+
// First event fires; the per-cluster cooldown then suppresses an immediate second event.
1012+
assertTrue(clusterDrsService.shouldTriggerEventDrivenDrs(1L));
1013+
assertFalse(clusterDrsService.shouldTriggerEventDrivenDrs(1L));
1014+
// A different cluster has an independent cooldown and still fires.
1015+
assertTrue(clusterDrsService.shouldTriggerEventDrivenDrs(2L));
1016+
} finally {
1017+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, origDrs);
1018+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, origEvt);
1019+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenInterval, origInt);
1020+
}
1021+
}
1022+
1023+
@Test
1024+
public void testTriggerEventDrivenDrsForVmSchedulesWhenEnabled() throws Exception {
1025+
String origDrs = getConfigDefault(clusterDrsService.ClusterDrsEnabled);
1026+
String origEvt = getConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled);
1027+
try {
1028+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, "true");
1029+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, "true");
1030+
1031+
VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
1032+
Mockito.when(vm.getHostId()).thenReturn(10L);
1033+
HostVO host = Mockito.mock(HostVO.class);
1034+
Mockito.when(host.getClusterId()).thenReturn(1L);
1035+
ClusterVO cluster = Mockito.mock(ClusterVO.class);
1036+
Mockito.when(vmInstanceDao.findById(100L)).thenReturn(vm);
1037+
Mockito.when(hostDao.findById(10L)).thenReturn(host);
1038+
Mockito.when(clusterDao.findById(1L)).thenReturn(cluster);
1039+
// Don't actually run DRS on a background thread in the test.
1040+
Mockito.doNothing().when(clusterDrsService).submitEventDrivenDrs(Mockito.any(ClusterVO.class), Mockito.anyInt());
1041+
1042+
clusterDrsService.triggerEventDrivenDrsForVm(100L);
1043+
1044+
Mockito.verify(clusterDrsService, Mockito.times(1)).submitEventDrivenDrs(Mockito.eq(cluster), Mockito.anyInt());
1045+
} finally {
1046+
setConfigDefault(clusterDrsService.ClusterDrsEnabled, origDrs);
1047+
setConfigDefault(clusterDrsService.ClusterDrsEventDrivenEnabled, origEvt);
1048+
}
1049+
}
1050+
1051+
@Test
1052+
public void testTriggerEventDrivenDrsForVmDoesNotScheduleWhenDisabled() {
1053+
// Defaults: both flags false -> must not schedule, even though the VM resolves to a cluster.
1054+
VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
1055+
Mockito.when(vm.getHostId()).thenReturn(10L);
1056+
HostVO host = Mockito.mock(HostVO.class);
1057+
Mockito.when(host.getClusterId()).thenReturn(1L);
1058+
Mockito.when(vmInstanceDao.findById(100L)).thenReturn(vm);
1059+
Mockito.when(hostDao.findById(10L)).thenReturn(host);
1060+
1061+
clusterDrsService.triggerEventDrivenDrsForVm(100L);
1062+
1063+
Mockito.verify(clusterDrsService, Mockito.never()).submitEventDrivenDrs(Mockito.any(ClusterVO.class), Mockito.anyInt());
1064+
}
9541065
}

0 commit comments

Comments
 (0)