CLOUDSTACK-9921: Fix NPE when storage garbage collector is running (#2139)

Steps to reproduce issue

Deploy a VM
Take snapshot of the root volume
Delete the snapshot
Before the garbage collector has run, shutdown the VM and assign the VM to other user.
When garage collector executes NPE shows in the logs.

engine/schema/src/com/cloud/storage/dao/SnapshotDao.java

@@ -61,4 +61,6 @@ public interface SnapshotDao extends GenericDao<SnapshotVO, Long>, StateDao<Snap
 
     void updateVolumeIds(long oldVolId, long newVolId);
 
+    List<SnapshotVO> listByStatusNotIn(long volumeId, Snapshot.State... status);
+
 }

engine/schema/src/com/cloud/storage/dao/SnapshotDaoImpl.java

@@ -69,6 +69,7 @@ public class SnapshotDaoImpl extends GenericDaoBase<SnapshotVO, Long> implements
     private SearchBuilder<SnapshotVO> AccountIdSearch;
     private SearchBuilder<SnapshotVO> InstanceIdSearch;
     private SearchBuilder<SnapshotVO> StatusSearch;
+    private SearchBuilder<SnapshotVO> notInStatusSearch;
     private GenericSearchBuilder<SnapshotVO, Long> CountSnapshotsByAccount;
     @Inject
     ResourceTagDao _tagsDao;
@@ -187,6 +188,11 @@ public class SnapshotDaoImpl extends GenericDaoBase<SnapshotVO, Long> implements
         StatusSearch.and("status", StatusSearch.entity().getState(), SearchCriteria.Op.IN);
         StatusSearch.done();
 
+        notInStatusSearch  = createSearchBuilder();
+        notInStatusSearch.and("volumeId", notInStatusSearch.entity().getVolumeId(), SearchCriteria.Op.EQ);
+        notInStatusSearch.and("status", notInStatusSearch.entity().getState(), SearchCriteria.Op.NOTIN);
+        notInStatusSearch.done();
+
         CountSnapshotsByAccount = createSearchBuilder(Long.class);
         CountSnapshotsByAccount.select(null, Func.COUNT, null);
         CountSnapshotsByAccount.and("account", CountSnapshotsByAccount.entity().getAccountId(), SearchCriteria.Op.EQ);
@@ -352,4 +358,12 @@ public class SnapshotDaoImpl extends GenericDaoBase<SnapshotVO, Long> implements
         UpdateBuilder ub = getUpdateBuilder(snapshot);
         update(ub, sc, null);
     }
+
+    @Override
+    public List<SnapshotVO> listByStatusNotIn(long volumeId, Snapshot.State... status) {
+        SearchCriteria<SnapshotVO> sc = this.notInStatusSearch.create();
+        sc.setParameters("volumeId", volumeId);
+        sc.setParameters("status", (Object[]) status);
+        return listBy(sc, null);
+    }
 }

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

@@ -231,10 +231,10 @@ import com.cloud.storage.GuestOSCategoryVO;
 import com.cloud.storage.GuestOSVO;
 import com.cloud.storage.SnapshotVO;
 import com.cloud.storage.Storage;
+import com.cloud.storage.Snapshot;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.Storage.TemplateType;
-import com.cloud.storage.Snapshot;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.StoragePoolStatus;
@@ -5457,11 +5457,20 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             }
         }
 
+        final List<VolumeVO> volumes = _volsDao.findByInstance(cmd.getVmId());
+
+        for (VolumeVO volume : volumes) {
+            List<SnapshotVO> snapshots = _snapshotDao.listByStatusNotIn(volume.getId(), Snapshot.State.Destroyed,Snapshot.State.Error);
+            if (snapshots != null && snapshots.size() > 0) {
+                throw new InvalidParameterValueException(
+                        "Snapshots exists for volume: "+ volume.getName()+ ", Detach volume or remove snapshots for volume before assigning VM to another user.");
+            }
+        }
+
         DataCenterVO zone = _dcDao.findById(vm.getDataCenterId());
 
         // Get serviceOffering and Volumes for Virtual Machine
         final ServiceOfferingVO offering = _serviceOfferingDao.findByIdIncludingRemoved(vm.getId(), vm.getServiceOfferingId());
-        final List<VolumeVO> volumes = _volsDao.findByInstance(cmd.getVmId());
 
         //Remove vm from instance group
         removeInstanceFromInstanceGroup(cmd.getVmId());
@@ -5516,16 +5525,6 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             _resourceLimitMgr.incrementResourceCount(newAccount.getAccountId(), ResourceType.primary_storage, new Long(volume.getSize()));
             UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_CREATE, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(),
                     volume.getDiskOfferingId(), volume.getTemplateId(), volume.getSize(), Volume.class.getName(), volume.getUuid(), volume.isDisplayVolume());
-            //snapshots: mark these removed in db
-            List<SnapshotVO> snapshots = _snapshotDao.listByVolumeIdIncludingRemoved(volume.getId());
-            for (SnapshotVO snapshot : snapshots) {
-                    boolean result = _snapshotService.deleteSnapshot(snapshot.getId());
-                    if (result) {
-                        s_logger.info("Snapshot id: " + snapshot.getId() + " delete successfully ");
-                    } else {
-                        s_logger.error("Unable to delete Snapshot id: " + snapshot.getId());
-                    }
-            }
         }
 
         //update resource count of new account

test/integration/component/test_assign_vm.py

@@ -112,6 +112,7 @@ class TestVMOwnership(cloudstackTestCase):
                                     cls.services["ostype"])
         cls.services["virtual_machine"]["zoneid"] = cls.zone.id
         cls.services["virtual_machine"]["template"] = cls.template.id
+        cls.hypervisor = cls.testClient.getHypervisorInfo()
 
         def create_domain_account_user(parentDomain=None):
             domain  =  Domain.create(cls.api_client,
@@ -415,16 +416,12 @@ class TestVMOwnership(cloudstackTestCase):
         """
         # Validate the following:
         # 1. deploy VM in sub subdomain1 with snapshot.
-        # 3. assignVirtualMachine to subdomain2
+        # 3. assign VM will fail with Exception as VM with snapshots cannot be assigned to other users.
         if self.hypervisor.lower() in ['hyperv', 'lxc']:
             self.skipTest("Snapshots feature is not supported on %s" % self.hypervisor)
         self.create_vm(self.sdomain_account_user1['account'], self.sdomain_account_user1['domain'], snapshot=True)
+        with self.assertRaises(Exception):
             self.virtual_machine.assign_virtual_machine(self.apiclient, self.sdomain_account_user2['account'].name ,self.sdomain_account_user2['domain'].id)
-        snapshots = list_snapshots(self.apiclient,
-                                   id=self.snapshot.id)
-        self.assertEqual(snapshots,
-                         None,
-                         "Snapshots stil present for a vm in domain")
 		
     @attr(tags = ["advanced"])
     @log_test_exceptions