Fix metrics stats for VMs not running

engine/schema/src/main/java/com/cloud/vm/dao/UserVmDao.java

@@ -60,6 +60,12 @@ public interface UserVmDao extends GenericDao<UserVmVO, Long> {
      */
     public List<UserVmVO> listRunningByHostId(long hostId);
 
+    /**
+     * List all running VMs.
+     * @return the list of all VM instances that are running.
+     */
+    public List<UserVmVO> listAllRunning();
+
     /**
      * List user vm instances with virtualized networking (i.e. not direct attached networking) for the given account and datacenter
      * @param accountId will search for vm instances belonging to this account

engine/schema/src/main/java/com/cloud/vm/dao/UserVmDaoImpl.java

@@ -67,6 +67,7 @@ public class UserVmDaoImpl extends GenericDaoBase<UserVmVO, Long> implements Use
     protected SearchBuilder<UserVmVO> LastHostSearch;
     protected SearchBuilder<UserVmVO> HostUpSearch;
     protected SearchBuilder<UserVmVO> HostRunningSearch;
+    protected SearchBuilder<UserVmVO> RunningSearch;
     protected SearchBuilder<UserVmVO> StateChangeSearch;
     protected SearchBuilder<UserVmVO> AccountHostSearch;
 
@@ -134,21 +135,22 @@ void init() {
         HostSearch.and("host", HostSearch.entity().getHostId(), SearchCriteria.Op.EQ);
         HostSearch.done();
 
-        LastHostSearch = createSearchBuilder();
+        LastHostSearch = createSearchBuilderWithStateCriteria(SearchCriteria.Op.EQ);
         LastHostSearch.and("lastHost", LastHostSearch.entity().getLastHostId(), SearchCriteria.Op.EQ);
-        LastHostSearch.and("state", LastHostSearch.entity().getState(), SearchCriteria.Op.EQ);
         LastHostSearch.done();
 
         HostUpSearch = createSearchBuilder();
         HostUpSearch.and("host", HostUpSearch.entity().getHostId(), SearchCriteria.Op.EQ);
         HostUpSearch.and("states", HostUpSearch.entity().getState(), SearchCriteria.Op.NIN);
         HostUpSearch.done();
 
-        HostRunningSearch = createSearchBuilder();
+        HostRunningSearch = createSearchBuilderWithStateCriteria(SearchCriteria.Op.EQ);
         HostRunningSearch.and("host", HostRunningSearch.entity().getHostId(), SearchCriteria.Op.EQ);
-        HostRunningSearch.and("state", HostRunningSearch.entity().getState(), SearchCriteria.Op.EQ);
         HostRunningSearch.done();
 
+        RunningSearch = createSearchBuilderWithStateCriteria(SearchCriteria.Op.EQ);
+        RunningSearch.done();
+
         AccountPodSearch = createSearchBuilder();
         AccountPodSearch.and("account", AccountPodSearch.entity().getAccountId(), SearchCriteria.Op.EQ);
         AccountPodSearch.and("pod", AccountPodSearch.entity().getPodIdToDeployIn(), SearchCriteria.Op.EQ);
@@ -209,6 +211,19 @@ void init() {
         assert _updateTimeAttr != null : "Couldn't get this updateTime attribute";
     }
 
+    /**
+     * Creates an {@link com.cloud.vm.UserVmVO UserVmVO} search builder with a
+     * {@link com.cloud.utils.db.SearchCriteria.Op SearchCriteria.Op} condition
+     * to the 'state' criteria already included.
+     * @param searchCriteria the {@link com.cloud.utils.db.SearchCriteria.Op SearchCriteria.Op} to 'state' criteria.
+     * @return the {@link com.cloud.vm.UserVmVO UserVmVO} search builder.
+     */
+    protected SearchBuilder<UserVmVO> createSearchBuilderWithStateCriteria(SearchCriteria.Op searchCriteria) {
+        SearchBuilder<UserVmVO> genericSearchBuilderWithStateCriteria = createSearchBuilder();
+        genericSearchBuilderWithStateCriteria.and("state", genericSearchBuilderWithStateCriteria.entity().getState(), searchCriteria);
+        return genericSearchBuilderWithStateCriteria;
+    }
+
     @Override
     public List<UserVmVO> listByAccountAndPod(long accountId, long podId) {
         SearchCriteria<UserVmVO> sc = AccountPodSearch.create();
@@ -298,6 +313,14 @@ public List<UserVmVO> listRunningByHostId(long hostId) {
         return listBy(sc);
     }
 
+    @Override
+    public List<UserVmVO> listAllRunning() {
+        SearchCriteria<UserVmVO> sc = RunningSearch.create();
+        sc.setParameters("state", State.Running);
+
+        return listBy(sc);
+    }
+
     @Override
     public List<UserVmVO> listVirtualNetworkInstancesByAcctAndNetwork(long accountId, long networkId) {
 

server/src/main/java/com/cloud/server/StatsCollector.java

@@ -238,7 +238,7 @@ public String toString() {
     @Inject
     private ClusterDao _clusterDao;
     @Inject
-    private UserVmDao _userVmDao;
+    protected UserVmDao _userVmDao;
     @Inject
     private VolumeDao _volsDao;
     @Inject
@@ -291,7 +291,7 @@ public String toString() {
     private ImageStoreDetailsUtil imageStoreDetailsUtil;
 
     private ConcurrentHashMap<Long, HostStats> _hostStats = new ConcurrentHashMap<Long, HostStats>();
-    private final ConcurrentHashMap<Long, VmStats> _VmStats = new ConcurrentHashMap<Long, VmStats>();
+    protected ConcurrentHashMap<Long, VmStats> _VmStats = new ConcurrentHashMap<Long, VmStats>();
     private final Map<String, VolumeStats> _volumeStats = new ConcurrentHashMap<String, VolumeStats>();
     private ConcurrentHashMap<Long, StorageStats> _storageStats = new ConcurrentHashMap<Long, StorageStats>();
     private ConcurrentHashMap<Long, StorageStats> _storagePoolStats = new ConcurrentHashMap<Long, StorageStats>();
@@ -619,6 +619,8 @@ protected void runInContext() {
                     }
                 }
 
+                cleanUpVirtualMachineStats();
+
             } catch (Throwable t) {
                 s_logger.error("Error trying to retrieve VM stats", t);
             }
@@ -1485,6 +1487,32 @@ private void storeVirtualMachineStatsInMemory(VmStatsEntry statsForCurrentIterat
         }
     }
 
+    /**
+     * Removes stats for a given virtual machine.
+     * @param vmId ID of the virtual machine to remove stats.
+     */
+    public void removeVirtualMachineStats(Long vmId) {
+        s_logger.debug(String.format("Removing stats from VM with ID: %s .",vmId));
+        _VmStats.remove(vmId);
+    }
+
+    /**
+     * Removes stats of virtual machines that are not running from memory.
+     */
+    protected void cleanUpVirtualMachineStats() {
+        List<Long> allRunningVmIds = new ArrayList<Long>();
+        for (UserVmVO vm : _userVmDao.listAllRunning()) {
+            allRunningVmIds.add(vm.getId());
+        }
+
+        List<Long> vmIdsToRemoveStats = new ArrayList<Long>(_VmStats.keySet());
+        vmIdsToRemoveStats.removeAll(allRunningVmIds);
+
+        for (Long vmId : vmIdsToRemoveStats) {
+            removeVirtualMachineStats(vmId);
+        }
+    }
+
     /**
      * Sends host metrics to a configured InfluxDB host. The metrics respects the following specification.</br>
      * <b>Tags:</b>vm_id, uuid, instance_name, data_center_id, host_id</br>

server/src/main/java/com/cloud/vm/UserVmManagerImpl.java

@@ -269,6 +269,7 @@
 import com.cloud.resource.ResourceState;
 import com.cloud.server.ManagementService;
 import com.cloud.server.ResourceTag;
+import com.cloud.server.StatsCollector;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.service.dao.ServiceOfferingDetailsDao;
@@ -552,6 +553,8 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     @Autowired
     @Qualifier("networkHelper")
     protected NetworkHelper nwHelper;
+    @Inject
+    private StatsCollector statsCollector;
 
     private ScheduledExecutorService _executor = null;
     private ScheduledExecutorService _vmIpFetchExecutor = null;
@@ -5001,6 +5004,8 @@ public UserVm stopVirtualMachine(long vmId, boolean forced) throws ConcurrentOpe
             throw new InvalidParameterValueException("unable to find a virtual machine with id " + vmId);
         }
 
+        statsCollector.removeVirtualMachineStats(vmId);
+
         _userDao.findById(userId);
         boolean status = false;
         try {
@@ -5306,6 +5311,8 @@ public UserVm destroyVm(long vmId, boolean expunge) throws ResourceUnavailableEx
             return vm;
         }
 
+        statsCollector.removeVirtualMachineStats(vmId);
+
         boolean status;
         State vmState = vm.getState();
 

server/src/test/java/com/cloud/server/StatsCollectorTest.java

@@ -21,9 +21,11 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 import org.influxdb.InfluxDB;
@@ -34,6 +36,8 @@
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PrepareForTest;
@@ -44,13 +48,18 @@
 import com.cloud.server.StatsCollector.ExternalStatsProtocol;
 import com.cloud.user.VmDiskStatisticsVO;
 import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.vm.UserVmVO;
+import com.cloud.vm.VmStats;
+import com.cloud.vm.dao.UserVmDao;
 import com.tngtech.java.junit.dataprovider.DataProvider;
 import com.tngtech.java.junit.dataprovider.DataProviderRunner;
 
 @RunWith(PowerMockRunner.class)
 @PowerMockRunnerDelegate(DataProviderRunner.class)
 @PrepareForTest({InfluxDBFactory.class, BatchPoints.class})
 public class StatsCollectorTest {
+
+    @InjectMocks
     private StatsCollector statsCollector = Mockito.spy(new StatsCollector());
 
     private static final int GRAPHITE_DEFAULT_PORT = 2003;
@@ -60,6 +69,18 @@ public class StatsCollectorTest {
 
     private static final String DEFAULT_DATABASE_NAME = "cloudstack";
 
+    @Mock
+    ConcurrentHashMap<Long, VmStats> vmStatsMock;
+
+    @Mock
+    VmStats singleVmStatsMock;
+
+    @Mock
+    UserVmDao userVmDaoMock;
+
+    @Mock
+    UserVmVO userVmVOMock;
+
     @Test
     public void createInfluxDbConnectionTest() {
         configureAndTestCreateInfluxDbConnection(true);
@@ -218,4 +239,48 @@ public void configureAndTestCheckIfDiskStatsAreZero(long bytesRead, long bytesWr
         boolean result = statsCollector.areAllDiskStatsZero(vmDiskStatsEntry);
         Assert.assertEquals(expected, result);
     }
+
+    @Test
+    public void removeVirtualMachineStatsTestRemoveOneVmStats() {
+        Mockito.doReturn(new Object()).when(vmStatsMock).remove(Mockito.anyLong());
+
+        statsCollector.removeVirtualMachineStats(1l);
+
+        Mockito.verify(vmStatsMock, Mockito.times(1)).remove(Mockito.anyLong());
+    }
+
+    @Test
+    public void cleanUpVirtualMachineStatsTestDoNothing() {
+        Mockito.doReturn(new ArrayList<>()).when(userVmDaoMock).listAllRunning();
+        Mockito.doReturn(new ConcurrentHashMap<Long, VmStats>(new HashMap<>()).keySet())
+        .when(vmStatsMock).keySet();
+
+        statsCollector.cleanUpVirtualMachineStats();
+
+        Mockito.verify(statsCollector, Mockito.never()).removeVirtualMachineStats(Mockito.anyLong());
+    }
+
+    @Test
+    public void cleanUpVirtualMachineStatsTestRemoveOneVmStats() {
+        Mockito.doReturn(new ArrayList<>()).when(userVmDaoMock).listAllRunning();
+        Mockito.doReturn(1l).when(userVmVOMock).getId();
+        Mockito.doReturn(new ConcurrentHashMap<Long, VmStats>(Map.of(1l, singleVmStatsMock)).keySet())
+        .when(vmStatsMock).keySet();
+
+        statsCollector.cleanUpVirtualMachineStats();
+
+        Mockito.verify(vmStatsMock, Mockito.times(1)).remove(Mockito.anyLong());
+    }
+
+    @Test
+    public void cleanUpVirtualMachineStatsTestRemoveOnlyOneVmStats() {
+        Mockito.doReturn(1l).when(userVmVOMock).getId();
+        Mockito.doReturn(Arrays.asList(userVmVOMock)).when(userVmDaoMock).listAllRunning();
+        Mockito.doReturn(new ConcurrentHashMap<Long, VmStats>(Map.of(1l, singleVmStatsMock, 2l, singleVmStatsMock)).keySet())
+        .when(vmStatsMock).keySet();
+
+        statsCollector.cleanUpVirtualMachineStats();
+
+        Mockito.verify(vmStatsMock, Mockito.times(1)).remove(Mockito.anyLong());
+    }
 }

server/src/test/java/com/cloud/user/AccountManagerImplVolumeDeleteEventTest.java

@@ -54,6 +54,7 @@
 import com.cloud.exception.AgentUnavailableException;
 import com.cloud.exception.CloudException;
 import com.cloud.exception.ConcurrentOperationException;
+import com.cloud.server.StatsCollector;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.storage.Volume.Type;
 import com.cloud.storage.VolumeVO;
@@ -75,6 +76,8 @@ public class AccountManagerImplVolumeDeleteEventTest extends AccountManagetImplT
     @Mock
     private UserVmManager userVmManager;
 
+    @Mock
+    private StatsCollector statsCollectorMock;
 
     Map<String, Object> oldFields = new HashMap<>();
     UserVmVO vm = mock(UserVmVO.class);
@@ -201,6 +204,7 @@ public void destroyedVMRootVolumeUsageEvent()
     // volume.
     public void runningVMRootVolumeUsageEvent()
             throws SecurityException, IllegalArgumentException, ReflectiveOperationException, AgentUnavailableException, ConcurrentOperationException, CloudException {
+        Mockito.doNothing().when(statsCollectorMock).removeVirtualMachineStats(Mockito.anyLong());
         Mockito.lenient().when(_vmMgr.destroyVm(nullable(Long.class), nullable(Boolean.class))).thenReturn(vm);
         List<UsageEventVO> emittedEvents = deleteUserAccountRootVolumeUsageEvents(false);
         UsageEventVO event = emittedEvents.get(0);