Make AgentManagerImpl.handleCommands resilient to listener exceptions

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<Integer, Listener> 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);
+            }
         }
     }
 

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));
+    }
 }

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(),

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<VMInstanceVO> sb = Mockito.mock(SearchBuilder.class);