Batch VM and host lookups in StorageManagerImpl to eliminate N+1 queries during cleanup and pool connect

cleanupStorage: replaced per-volume _vmInstanceDao.findById(vol.getInstanceId())
with a single batch SELECT ... WHERE id IN (...) before the loop, using a
HashMap for O(1) lookups. Added null guard for volumes with null instanceId.

connectHostsToPool: replaced per-thread _hostDao.findById(hostId) with a
batch load before thread pool submission. Each thread now reads from a
pre-built HashMap instead of hitting the DB. Added null guard with warning
log for hosts removed between list assembly and batch read.

Added VMInstanceDao.listByIds(List<Long>) with empty-list guard and IN-clause
SearchBuilder, matching the established pattern in HostDao.

Author: sliceofapplepie

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

@@ -199,4 +199,6 @@ List<VMInstanceVO> searchRemovedByRemoveDate(final Date startDate, final Date en
     List<VMInstanceVO> listDeleteProtectedVmsByDomainIds(Set<Long> domainIds);
 
     List<Long> listIdsByHostIdForVolumeStats(long hostIds);
+
+    List<VMInstanceVO> listByIds(List<Long> ids);
 }

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

@@ -1352,4 +1352,17 @@ public List<Long> listIdsByHostIdForVolumeStats(long hostId) {
         sc.setParameters("lastHost", hostId);
         return customSearch(sc, null);
     }
+
+    @Override
+    public List<VMInstanceVO> listByIds(List<Long> ids) {
+        if (ids == null || ids.isEmpty()) {
+            return new ArrayList<>();
+        }
+        SearchBuilder<VMInstanceVO> sb = createSearchBuilder();
+        sb.and("id", sb.entity().getId(), Op.IN);
+        sb.done();
+        SearchCriteria<VMInstanceVO> sc = sb.create();
+        sc.setParameters("id", ids.toArray());
+        return listBy(sc);
+    }
 }

server/src/main/java/com/cloud/storage/StorageManagerImpl.java

@@ -1846,6 +1846,10 @@ public void connectHostsToPool(DataStore primaryStore, List<Long> hostIds, Scope
         if (CollectionUtils.isEmpty(hostIds)) {
             return;
         }
+        Map<Long, HostVO> hostMap = new HashMap<>();
+        for (HostVO h : _hostDao.listByIds(hostIds)) {
+            hostMap.put(h.getId(), h);
+        }
         CopyOnWriteArrayList<Long> poolHostIds = new CopyOnWriteArrayList<>();
         ExecutorService executorService = Executors.newFixedThreadPool(Math.max(1, Math.min(hostIds.size(),
                 StoragePoolHostConnectWorkers.value())));
@@ -1856,10 +1860,14 @@ public void connectHostsToPool(DataStore primaryStore, List<Long> hostIds, Scope
                 if (exceptionOccurred.get()) {
                     return null;
                 }
+                HostVO host = hostMap.get(hostId);
+                if (host == null) {
+                    logger.warn("Host [{}] not found, skipping pool connection for {}", hostId, primaryStore);
+                    return null;
+                }
                 Transaction.execute(new TransactionCallbackWithExceptionNoReturn<Exception>() {
                     @Override
                     public void doInTransactionWithoutResult(TransactionStatus status) throws Exception {
-                        HostVO host = _hostDao.findById(hostId);
                         try {
                             connectHostToSharedPool(host, primaryStore.getId());
                             poolHostIds.add(hostId);
@@ -2126,9 +2134,18 @@ public void cleanupStorage(boolean recurring) {
                     cleanupSecondaryStorage(recurring);
 
                     List<VolumeVO> vols = volumeDao.listVolumesToBeDestroyed(new Date(System.currentTimeMillis() - ((long)StorageCleanupDelay.value() << 10)));
+                    List<Long> rootVolumeVmIds = vols.stream()
+                            .filter(v -> Type.ROOT.equals(v.getVolumeType()) && v.getInstanceId() != null)
+                            .map(VolumeVO::getInstanceId)
+                            .distinct()
+                            .collect(Collectors.toList());
+                    Map<Long, VMInstanceVO> vmMap = new HashMap<>();
+                    for (VMInstanceVO vm : _vmInstanceDao.listByIds(rootVolumeVmIds)) {
+                        vmMap.put(vm.getId(), vm);
+                    }
                     for (VolumeVO vol : vols) {
                         if (Type.ROOT.equals(vol.getVolumeType())) {
-                             VMInstanceVO vmInstanceVO = _vmInstanceDao.findById(vol.getInstanceId());
+                            VMInstanceVO vmInstanceVO = vol.getInstanceId() != null ? vmMap.get(vol.getInstanceId()) : null;
                              if (vmInstanceVO != null && vmInstanceVO.getState() == State.Destroyed) {
                                  logger.debug("ROOT volume [{}] will not be expunged because the VM is [{}], therefore this volume will be expunged with the VM"
                                          + " cleanup job.", vol, vmInstanceVO.getState());