diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java index 1215829d92f8..aace076ad8d9 100644 --- a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java +++ b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java @@ -456,8 +456,13 @@ private AgentControlAnswer handleControlCommand(final AgentAttache attache, fina public void handleCommands(final AgentAttache attache, final long sequence, final Command[] cmds) { for (final Pair listener : _cmdMonitors) { - final boolean processed = listener.second().processCommands(attache.getId(), sequence, cmds); - logger.trace("SeqA {}-{}: {} by {}", attache.getId(), sequence, (processed ? "processed" : "not processed"), listener.getClass()); + try { + final boolean processed = listener.second().processCommands(attache.getId(), sequence, cmds); + logger.trace("SeqA {}-{}: {} by {}", attache.getId(), sequence, (processed ? "processed" : "not processed"), listener.second().getClass()); + } catch (final Exception e) { + logger.warn("Listener {} threw an exception processing commands for agent {} seq {}; continuing to next listener", + listener.second().getClass().getName(), attache.getId(), sequence, e); + } } } diff --git a/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java b/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java index 43d83a672c0f..d5b698f4c479 100644 --- a/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java +++ b/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java @@ -18,6 +18,7 @@ import com.cloud.agent.Listener; import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; import com.cloud.agent.api.ReadyCommand; import com.cloud.agent.api.StartupCommand; import com.cloud.agent.api.StartupRoutingCommand; @@ -137,4 +138,84 @@ public void testGetHostSshPortWithKVMHostCustomPort() { int hostSshPort = mgr.getHostSshPort(host); Assert.assertEquals(3922, hostSshPort); } + + /* + * When the first registered listener throws a RuntimeException, + * handleCommands must catch it, log a warning, and continue so that + * the second listener still runs. + */ + @Test + public void testHandleCommandsListenerExceptionDoesNotAbortLoop() throws Exception { + Listener throwingListener = Mockito.mock(Listener.class); + Mockito.when(throwingListener.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenThrow(new RuntimeException("simulated NPE from power-state sync")); + + Listener goodListener = Mockito.mock(Listener.class); + Mockito.when(goodListener.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenReturn(true); + + mgr._cmdMonitors = new ArrayList<>(); + mgr._cmdMonitors.add(new Pair<>(1, throwingListener)); + mgr._cmdMonitors.add(new Pair<>(2, goodListener)); + + Command[] cmds = new Command[]{Mockito.mock(Command.class)}; + + // Must not throw; the exception from throwingListener must be swallowed. + mgr.handleCommands(attache, 1L, cmds); + + // The second listener must still have been invoked. + Mockito.verify(goodListener, Mockito.times(1)) + .processCommands(Mockito.eq(attache.getId()), Mockito.eq(1L), Mockito.eq(cmds)); + } + + @Test + public void testHandleCommandsAllListenersSucceed() throws Exception { + Listener listenerA = Mockito.mock(Listener.class); + Mockito.when(listenerA.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenReturn(true); + + Listener listenerB = Mockito.mock(Listener.class); + Mockito.when(listenerB.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenReturn(false); + + mgr._cmdMonitors = new ArrayList<>(); + mgr._cmdMonitors.add(new Pair<>(1, listenerA)); + mgr._cmdMonitors.add(new Pair<>(2, listenerB)); + + Command[] cmds = new Command[]{Mockito.mock(Command.class)}; + mgr.handleCommands(attache, 1L, cmds); + + Mockito.verify(listenerA, Mockito.times(1)) + .processCommands(Mockito.eq(attache.getId()), Mockito.eq(1L), Mockito.eq(cmds)); + Mockito.verify(listenerB, Mockito.times(1)) + .processCommands(Mockito.eq(attache.getId()), Mockito.eq(1L), Mockito.eq(cmds)); + } + + /* + * Simulates the reported failure: a power-state sync listener throws + * partway through, and a downstream ping-style listener must still run + * so pingBy() isn't starved. + */ + @Test + public void testHandleCommandsThrowingListenerDoesNotStarveDownstreamListener() throws Exception { + Listener powerStateSyncListener = Mockito.mock(Listener.class); + Mockito.when(powerStateSyncListener.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenThrow(new NullPointerException("hostId was null in isPowerStateInSyncWithInstanceState")); + + Listener pingListener = Mockito.mock(Listener.class); + Mockito.when(pingListener.processCommands(Mockito.anyLong(), Mockito.anyLong(), Mockito.any())) + .thenReturn(false); + + mgr._cmdMonitors = new ArrayList<>(); + mgr._cmdMonitors.add(new Pair<>(1, powerStateSyncListener)); + mgr._cmdMonitors.add(new Pair<>(2, pingListener)); + + Command[] cmds = new Command[]{Mockito.mock(Command.class)}; + + // Before the fix this would abort on the NPE and pingListener would never run. + mgr.handleCommands(attache, 42L, cmds); + + Mockito.verify(pingListener, Mockito.times(1)) + .processCommands(Mockito.eq(attache.getId()), Mockito.eq(42L), Mockito.eq(cmds)); + } } diff --git a/engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDaoImpl.java b/engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDaoImpl.java index ae1e838649ba..1d629f68c3af 100755 --- a/engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDaoImpl.java +++ b/engine/schema/src/main/java/com/cloud/vm/dao/VMInstanceDaoImpl.java @@ -355,6 +355,7 @@ protected void init() { IdsPowerStateSelectSearch = createSearchBuilder(); IdsPowerStateSelectSearch.and("id", IdsPowerStateSelectSearch.entity().getId(), Op.IN); IdsPowerStateSelectSearch.selectFields(IdsPowerStateSelectSearch.entity().getId(), + IdsPowerStateSelectSearch.entity().getHostId(), IdsPowerStateSelectSearch.entity().getPowerHostId(), IdsPowerStateSelectSearch.entity().getPowerState(), IdsPowerStateSelectSearch.entity().getPowerStateUpdateCount(), diff --git a/engine/schema/src/test/java/com/cloud/vm/dao/VMInstanceDaoImplTest.java b/engine/schema/src/test/java/com/cloud/vm/dao/VMInstanceDaoImplTest.java index 5f8b2dd90eca..8d7a1e86a59f 100644 --- a/engine/schema/src/test/java/com/cloud/vm/dao/VMInstanceDaoImplTest.java +++ b/engine/schema/src/test/java/com/cloud/vm/dao/VMInstanceDaoImplTest.java @@ -210,6 +210,53 @@ public void testUpdatePowerStateNoChangeMaxUpdatesInvalidStateVmRunning() { assertTrue(result); } + /* + * Mirrors testUpdatePowerStateNoChangeMaxUpdatesInvalidStateVmStopped but with a null + * hostId (the default from setUp), which is what happens when the row was loaded through + * a partial-select projection that doesn't include hostId. isPowerStateInSyncWithInstanceState + * must not NPE on the null hostId, and must still detect the state is out-of-sync. + */ + @Test + public void testUpdatePowerStateNoChangeMaxUpdatesInvalidStateVmStoppedNullHostId() { + when(vm.getPowerStateUpdateTime()).thenReturn(null); + when(vm.getPowerHostId()).thenReturn(1L); + when(vm.getPowerState()).thenReturn(VirtualMachine.PowerState.PowerOn); + when(vm.getState()).thenReturn(Stopped); + doReturn(vm).when(vmInstanceDao).findById(anyLong()); + doReturn(true).when(vmInstanceDao).update(anyLong(), any()); + + boolean result = vmInstanceDao.updatePowerState(1L, 1L, VirtualMachine.PowerState.PowerOn, new Date()); + + verify(vm, times(1)).setPowerState(any()); + verify(vm, times(1)).setPowerHostId(anyLong()); + verify(vm, times(1)).setPowerStateUpdateCount(1); + verify(vm, times(1)).setPowerStateUpdateTime(any(Date.class)); + + assertTrue(result); + } + + /* + * Mirrors testUpdatePowerStateNoChangeMaxUpdatesInvalidStateVmRunning but with a null hostId. + */ + @Test + public void testUpdatePowerStateNoChangeMaxUpdatesInvalidStateVmRunningNullHostId() { + when(vm.getPowerStateUpdateTime()).thenReturn(null); + when(vm.getPowerHostId()).thenReturn(1L); + when(vm.getPowerState()).thenReturn(VirtualMachine.PowerState.PowerOff); + when(vm.getState()).thenReturn(Running); + doReturn(vm).when(vmInstanceDao).findById(anyLong()); + doReturn(true).when(vmInstanceDao).update(anyLong(), any()); + + boolean result = vmInstanceDao.updatePowerState(1L, 1L, VirtualMachine.PowerState.PowerOff, new Date()); + + verify(vm, times(1)).setPowerState(any()); + verify(vm, times(1)).setPowerHostId(anyLong()); + verify(vm, times(1)).setPowerStateUpdateCount(1); + verify(vm, times(1)).setPowerStateUpdateTime(any(Date.class)); + + assertTrue(result); + } + @Test public void testSearchRemovedByRemoveDate() { SearchBuilder sb = Mockito.mock(SearchBuilder.class);