KVM: Option to deploy a VM with existing volume/snapshot (#10503)

* Option to deploy a VM with existing volume/snapshot

* smoke test changes

check if the hypervisor is KVM
check if the primary storage's scope is ZONE wide

* skip all tests if the storage isn't Zone-Wide and the hypervisor isn't KVM

* support StorPool tags

add StorPool tags to a volume created from snapshot or to a volume which
will be attached as a ROOT to a new VM

* Add StorPool tags on the new ROOT volume

* Add the StorPool's tags when volume is created from a snapshot or a
volume is attached as a ROOT to a VM

* Addressed review

api/src/main/java/com/cloud/vm/UserVmService.java

@@ -16,6 +16,8 @@
 // under the License.
 package com.cloud.vm;
 
+import com.cloud.storage.Snapshot;
+import com.cloud.storage.Volume;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -222,7 +224,7 @@ public interface UserVmService {
                                                   String userData, Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIp, Boolean displayVm, String keyboard,
                                                   List<Long> affinityGroupIdList, Map<String, String> customParameter, String customId, Map<String, Map<Integer, String>> dhcpOptionMap,
                                                   Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                                  Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId) throws InsufficientCapacityException,
+                                                  Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException,
         ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException;
 
     /**
@@ -298,7 +300,7 @@ public interface UserVmService {
                                                      List<Long> securityGroupIdList, Account owner, String hostName, String displayName, Long diskOfferingId, Long diskSize, String group, HypervisorType hypervisor,
                                                      HTTPMethod httpmethod, String userData, Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean displayVm, String keyboard,
                                                      List<Long> affinityGroupIdList, Map<String, String> customParameters, String customId, Map<String, Map<Integer, String>> dhcpOptionMap,
-                                                     Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, String vmType) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException;
+                                                     Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, String vmType, Volume volume, Snapshot snapshot) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException;
 
     /**
      * Creates a User VM in Advanced Zone (Security Group feature is disabled)
@@ -370,7 +372,7 @@ public interface UserVmService {
                                         String hostName, String displayName, Long diskOfferingId, Long diskSize, String group, HypervisorType hypervisor, HTTPMethod httpmethod, String userData,
                                         Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean displayVm, String keyboard, List<Long> affinityGroupIdList,
                                         Map<String, String> customParameters, String customId, Map<String, Map<Integer, String>> dhcpOptionMap, Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                        Map<String, String> templateOvfPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId)
+                                        Map<String, String> templateOvfPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId, Volume volume, Snapshot snapshot)
 
         throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException;
 

api/src/main/java/org/apache/cloudstack/api/command/user/vm/DeployVMCmd.java

@@ -30,10 +30,14 @@ import com.cloud.network.Network.IpAddresses;
 import com.cloud.offering.DiskOffering;
 import com.cloud.template.VirtualMachineTemplate;
 import com.cloud.uservm.UserVm;
+import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.net.Dhcp;
 import com.cloud.utils.net.NetUtils;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VmDetailConstants;
+
+import java.util.Objects;
+import java.util.stream.Stream;
 import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.affinity.AffinityGroupResponse;
 import org.apache.cloudstack.api.ACL;
@@ -55,9 +59,11 @@ import org.apache.cloudstack.api.response.NetworkResponse;
 import org.apache.cloudstack.api.response.ProjectResponse;
 import org.apache.cloudstack.api.response.SecurityGroupResponse;
 import org.apache.cloudstack.api.response.ServiceOfferingResponse;
+import org.apache.cloudstack.api.response.SnapshotResponse;
 import org.apache.cloudstack.api.response.TemplateResponse;
 import org.apache.cloudstack.api.response.UserDataResponse;
 import org.apache.cloudstack.api.response.UserVmResponse;
+import org.apache.cloudstack.api.response.VolumeResponse;
 import org.apache.cloudstack.api.response.ZoneResponse;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.vm.lease.VMLeaseManager;
@@ -95,7 +101,7 @@ public class DeployVMCmd extends BaseAsyncCreateCustomIdCmd implements SecurityG
     private Long serviceOfferingId;
 
     @ACL
-    @Parameter(name = ApiConstants.TEMPLATE_ID, type = CommandType.UUID, entityType = TemplateResponse.class, required = true, description = "the ID of the template for the virtual machine")
+    @Parameter(name = ApiConstants.TEMPLATE_ID, type = CommandType.UUID, entityType = TemplateResponse.class, description = "the ID of the template for the virtual machine")
     private Long templateId;
 
     @Parameter(name = ApiConstants.NAME, type = CommandType.STRING, description = "host name for the virtual machine", validations = {ApiArgValidator.RFCComplianceDomainName})
@@ -286,6 +292,11 @@ public class DeployVMCmd extends BaseAsyncCreateCustomIdCmd implements SecurityG
             description = "Lease expiry action, valid values are STOP and DESTROY")
     private String leaseExpiryAction;
 
+    @Parameter(name = ApiConstants.VOLUME_ID, type = CommandType.UUID, entityType = VolumeResponse.class, since = "4.21")
+    private Long volumeId;
+
+    @Parameter(name = ApiConstants.SNAPSHOT_ID, type = CommandType.UUID, entityType = SnapshotResponse.class, since = "4.21")
+    private Long snapshotId;
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -744,6 +755,18 @@ public class DeployVMCmd extends BaseAsyncCreateCustomIdCmd implements SecurityG
         }
         return null;
     }
+
+    public Long getVolumeId() {
+        return volumeId;
+    }
+
+    public Long getSnapshotId() {
+        return snapshotId;
+    }
+
+    public boolean isVolumeOrSnapshotProvided() {
+        return volumeId != null || snapshotId != null;
+    }
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -840,6 +863,10 @@ public class DeployVMCmd extends BaseAsyncCreateCustomIdCmd implements SecurityG
 
     @Override
     public void create() throws ResourceAllocationException {
+        if (Stream.of(templateId, snapshotId, volumeId).filter(Objects::nonNull).count() != 1) {
+            throw new CloudRuntimeException("Please provide only one of the following parameters - template ID, volume ID or snapshot ID");
+        }
+
         try {
             UserVm vm = _userVmService.createVirtualMachine(this);
 

engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java

@@ -16,6 +16,8 @@
 // under the License.
 package com.cloud.vm;
 
+import com.cloud.storage.Snapshot;
+import com.cloud.storage.Volume;
 import java.net.URI;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -129,11 +131,11 @@ public interface VirtualMachineManager extends Manager {
      * @throws InsufficientCapacityException If there are insufficient capacity to deploy this vm.
      */
     void allocate(String vmInstanceName, VirtualMachineTemplate template, ServiceOffering serviceOffering, DiskOfferingInfo rootDiskOfferingInfo,
-        List<DiskOfferingInfo> dataDiskOfferings, LinkedHashMap<? extends Network, List<? extends NicProfile>> auxiliaryNetworks, DeploymentPlan plan,
+                  List<DiskOfferingInfo> dataDiskOfferings, LinkedHashMap<? extends Network, List<? extends NicProfile>> auxiliaryNetworks, DeploymentPlan plan,
-        HypervisorType hyperType, Map<String, Map<Integer, String>> extraDhcpOptions, Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap) throws InsufficientCapacityException;
+                  HypervisorType hyperType, Map<String, Map<Integer, String>> extraDhcpOptions, Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap, Volume volume, Snapshot snapshot) throws InsufficientCapacityException;
 
     void allocate(String vmInstanceName, VirtualMachineTemplate template, ServiceOffering serviceOffering,
-        LinkedHashMap<? extends Network, List<? extends NicProfile>> networkProfiles, DeploymentPlan plan, HypervisorType hyperType) throws InsufficientCapacityException;
+        LinkedHashMap<? extends Network, List<? extends NicProfile>> networkProfiles, DeploymentPlan plan, HypervisorType hyperType, Volume volume, Snapshot snapshot) throws InsufficientCapacityException;
 
     void start(String vmUuid, Map<VirtualMachineProfile.Param, Object> params);
 

engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/VolumeOrchestrationService.java

@@ -149,7 +149,7 @@ public interface VolumeOrchestrationService {
      * Allocate a volume or multiple volumes in case of template is registered with the 'deploy-as-is' option, allowing multiple disks
      */
     List<DiskProfile> allocateTemplatedVolumes(Type type, String name, DiskOffering offering, Long rootDisksize, Long minIops, Long maxIops, VirtualMachineTemplate template, VirtualMachine vm,
-                                               Account owner);
+                                               Account owner, Volume volume, Snapshot snapshot);
 
     String getVmNameFromVolumeId(long volumeId);
 

engine/api/src/main/java/org/apache/cloudstack/engine/service/api/OrchestrationService.java

@@ -18,6 +18,8 @@
  */
 package org.apache.cloudstack.engine.service.api;
 
+import com.cloud.storage.Snapshot;
+import com.cloud.storage.Volume;
 import java.net.URL;
 import java.util.List;
 import java.util.Map;
@@ -62,12 +64,12 @@ public interface OrchestrationService {
     @POST
     @Path("/createvm")
     VirtualMachineEntity createVirtualMachine(@QueryParam("id") String id, @QueryParam("owner") String owner, @QueryParam("template-id") String templateId,
-        @QueryParam("host-name") String hostName, @QueryParam("display-name") String displayName, @QueryParam("hypervisor") String hypervisor,
+                                              @QueryParam("host-name") String hostName, @QueryParam("display-name") String displayName, @QueryParam("hypervisor") String hypervisor,
-        @QueryParam("cpu") int cpu, @QueryParam("speed") int speed, @QueryParam("ram") long memory, @QueryParam("disk-size") Long diskSize,
+                                              @QueryParam("cpu") int cpu, @QueryParam("speed") int speed, @QueryParam("ram") long memory, @QueryParam("disk-size") Long diskSize,
-        @QueryParam("compute-tags") List<String> computeTags, @QueryParam("root-disk-tags") List<String> rootDiskTags,
+                                              @QueryParam("compute-tags") List<String> computeTags, @QueryParam("root-disk-tags") List<String> rootDiskTags,
-        @QueryParam("network-nic-map") Map<String, List<NicProfile>> networkNicMap, @QueryParam("deploymentplan") DeploymentPlan plan,
+                                              @QueryParam("network-nic-map") Map<String, List<NicProfile>> networkNicMap, @QueryParam("deploymentplan") DeploymentPlan plan,
-        @QueryParam("root-disk-size") Long rootDiskSize, @QueryParam("extra-dhcp-option-map") Map<String, Map<Integer, String>> extraDhcpOptionMap,
+                                              @QueryParam("root-disk-size") Long rootDiskSize, @QueryParam("extra-dhcp-option-map") Map<String, Map<Integer, String>> extraDhcpOptionMap,
-        @QueryParam("datadisktemplate-diskoffering-map") Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap, @QueryParam("disk-offering-id") Long diskOfferingId, @QueryParam("root-disk-offering-id") Long rootDiskOfferingId) throws InsufficientCapacityException;
+                                              @QueryParam("datadisktemplate-diskoffering-map") Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap, @QueryParam("disk-offering-id") Long diskOfferingId, @QueryParam("root-disk-offering-id") Long rootDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException;
 
     @POST
     VirtualMachineEntity createVirtualMachineFromScratch(@QueryParam("id") String id, @QueryParam("owner") String owner, @QueryParam("iso-id") String isoId,
@@ -75,7 +77,7 @@ public interface OrchestrationService {
         @QueryParam("os") String os, @QueryParam("cpu") int cpu, @QueryParam("speed") int speed, @QueryParam("ram") long memory, @QueryParam("disk-size") Long diskSize,
         @QueryParam("compute-tags") List<String> computeTags, @QueryParam("root-disk-tags") List<String> rootDiskTags,
         @QueryParam("network-nic-map") Map<String, List<NicProfile>> networkNicMap, @QueryParam("deploymentplan") DeploymentPlan plan,
-        @QueryParam("extra-dhcp-option-map") Map<String,  Map<Integer, String>> extraDhcpOptionMap, @QueryParam("disk-offering-id") Long diskOfferingId) throws InsufficientCapacityException;
+        @QueryParam("extra-dhcp-option-map") Map<String,  Map<Integer, String>> extraDhcpOptionMap, @QueryParam("disk-offering-id") Long diskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException;
 
     @POST
     NetworkEntity createNetwork(String id, String name, String domainName, String cidr, String gateway);

engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java

@@ -49,6 +49,7 @@ import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 import javax.persistence.EntityExistsException;
 
+
 import org.apache.cloudstack.affinity.dao.AffinityGroupVMMapDao;
 import org.apache.cloudstack.annotation.AnnotationService;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
@@ -230,6 +231,7 @@ import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.ScopeType;
+import com.cloud.storage.Snapshot;
 import com.cloud.storage.Storage;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.StorageManager;
@@ -291,6 +293,7 @@ import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
 import com.google.gson.Gson;
 
+
 public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMachineManager, VmWorkJobHandler, Listener, Configurable {
 
     public static final String VM_WORK_JOB_HANDLER = VirtualMachineManagerImpl.class.getSimpleName();
@@ -503,8 +506,8 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
     @Override
     @DB
     public void allocate(final String vmInstanceName, final VirtualMachineTemplate template, final ServiceOffering serviceOffering,
-            final DiskOfferingInfo rootDiskOfferingInfo, final List<DiskOfferingInfo> dataDiskOfferings,
+                         final DiskOfferingInfo rootDiskOfferingInfo, final List<DiskOfferingInfo> dataDiskOfferings,
-            final LinkedHashMap<? extends Network, List<? extends NicProfile>> auxiliaryNetworks, final DeploymentPlan plan, final HypervisorType hyperType, final Map<String, Map<Integer, String>> extraDhcpOptions, final Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap)
+                         final LinkedHashMap<? extends Network, List<? extends NicProfile>> auxiliaryNetworks, final DeploymentPlan plan, final HypervisorType hyperType, final Map<String, Map<Integer, String>> extraDhcpOptions, final Map<Long, DiskOffering> datadiskTemplateToDiskOfferingMap, Volume volume, Snapshot snapshot)
                     throws InsufficientCapacityException {
 
         logger.info("allocating virtual machine from template: {} with hostname: {} and {} networks", template, vmInstanceName, auxiliaryNetworks.size());
@@ -542,7 +545,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
 
             logger.debug("Allocating disks for {}",  persistedVm);
 
-            allocateRootVolume(persistedVm, template, rootDiskOfferingInfo, owner, rootDiskSizeFinal);
+            allocateRootVolume(persistedVm, template, rootDiskOfferingInfo, owner, rootDiskSizeFinal, volume, snapshot);
 
             // Create new Volume context and inject event resource type, id and details to generate VOLUME.CREATE event for the ROOT disk.
             CallContext volumeContext = CallContext.register(CallContext.current(), ApiCommandResourceType.Volume);
@@ -583,7 +586,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
         }
     }
 
-    private void allocateRootVolume(VMInstanceVO vm, VirtualMachineTemplate template, DiskOfferingInfo rootDiskOfferingInfo, Account owner, Long rootDiskSizeFinal) {
+    private void allocateRootVolume(VMInstanceVO vm, VirtualMachineTemplate template, DiskOfferingInfo rootDiskOfferingInfo, Account owner, Long rootDiskSizeFinal, Volume volume, Snapshot snapshot) {
         // Create new Volume context and inject event resource type, id and details to generate VOLUME.CREATE event for the ROOT disk.
         CallContext volumeContext = CallContext.register(CallContext.current(), ApiCommandResourceType.Volume);
         try {
@@ -595,7 +598,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
                 logger.debug("%s has format [{}]. Skipping ROOT volume [{}] allocation.", template.toString(), ImageFormat.BAREMETAL, rootVolumeName);
             } else {
                 volumeMgr.allocateTemplatedVolumes(Type.ROOT, rootVolumeName, rootDiskOfferingInfo.getDiskOffering(), rootDiskSizeFinal,
-                        rootDiskOfferingInfo.getMinIops(), rootDiskOfferingInfo.getMaxIops(), template, vm, owner);
+                        rootDiskOfferingInfo.getMinIops(), rootDiskOfferingInfo.getMaxIops(), template, vm, owner, volume, snapshot);
             }
         } finally {
             // Remove volumeContext and pop vmContext back
@@ -605,9 +608,9 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
 
     @Override
     public void allocate(final String vmInstanceName, final VirtualMachineTemplate template, final ServiceOffering serviceOffering,
-            final LinkedHashMap<? extends Network, List<? extends NicProfile>> networks, final DeploymentPlan plan, final HypervisorType hyperType) throws InsufficientCapacityException {
+            final LinkedHashMap<? extends Network, List<? extends NicProfile>> networks, final DeploymentPlan plan, final HypervisorType hyperType, Volume volume, Snapshot snapshot) throws InsufficientCapacityException {
         DiskOffering diskOffering = _diskOfferingDao.findById(serviceOffering.getDiskOfferingId());
-        allocate(vmInstanceName, template, serviceOffering, new DiskOfferingInfo(diskOffering), new ArrayList<>(), networks, plan, hyperType, null, null);
+        allocate(vmInstanceName, template, serviceOffering, new DiskOfferingInfo(diskOffering), new ArrayList<>(), networks, plan, hyperType, null, null, volume, snapshot);
     }
 
     VirtualMachineGuru getVmGuru(final VirtualMachine vm) {

engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/CloudOrchestrator.java

@@ -18,6 +18,9 @@
  */
 package org.apache.cloudstack.engine.orchestration;
 
+import com.cloud.storage.Snapshot;
+import com.cloud.storage.Volume;
+import com.cloud.template.VirtualMachineTemplate;
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
@@ -158,8 +161,8 @@ public class CloudOrchestrator implements OrchestrationService {
 
     @Override
     public VirtualMachineEntity createVirtualMachine(String id, String owner, String templateId, String hostName, String displayName, String hypervisor, int cpu,
-        int speed, long memory, Long diskSize, List<String> computeTags, List<String> rootDiskTags, Map<String, List<NicProfile>> networkNicMap, DeploymentPlan plan,
+                                                     int speed, long memory, Long diskSize, List<String> computeTags, List<String> rootDiskTags, Map<String, List<NicProfile>> networkNicMap, DeploymentPlan plan,
-        Long rootDiskSize, Map<String, Map<Integer, String>> extraDhcpOptionMap, Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Long dataDiskOfferingId, Long rootDiskOfferingId) throws InsufficientCapacityException {
+                                                     Long rootDiskSize, Map<String, Map<Integer, String>> extraDhcpOptionMap, Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Long dataDiskOfferingId, Long rootDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException {
 
         // VirtualMachineEntityImpl vmEntity = new VirtualMachineEntityImpl(id, owner, hostName, displayName, cpu, speed, memory, computeTags, rootDiskTags, networks,
         // vmEntityManager);
@@ -254,9 +257,13 @@ public class CloudOrchestrator implements OrchestrationService {
                 }
             }
         }
-
+        VirtualMachineTemplate template = null;
-        _itMgr.allocate(vm.getInstanceName(), _templateDao.findById(new Long(templateId)), computeOffering, rootDiskOfferingInfo, dataDiskOfferings, networkIpMap, plan,
+        if (volume != null || snapshot != null) {
-            hypervisorType, extraDhcpOptionMap, dataDiskTemplateToDiskOfferingMap);
+            template = _templateDao.findByIdIncludingRemoved(new Long(templateId));
+        } else
+            template = _templateDao.findById(new Long(templateId));
+        _itMgr.allocate(vm.getInstanceName(), template, computeOffering, rootDiskOfferingInfo, dataDiskOfferings, networkIpMap, plan,
+            hypervisorType, extraDhcpOptionMap, dataDiskTemplateToDiskOfferingMap, volume, snapshot);
 
         return vmEntity;
     }
@@ -264,7 +271,7 @@ public class CloudOrchestrator implements OrchestrationService {
     @Override
     public VirtualMachineEntity createVirtualMachineFromScratch(String id, String owner, String isoId, String hostName, String displayName, String hypervisor, String os,
         int cpu, int speed, long memory, Long diskSize, List<String> computeTags, List<String> rootDiskTags, Map<String, List<NicProfile>> networkNicMap, DeploymentPlan plan,
-        Map<String, Map<Integer, String>> extraDhcpOptionMap, Long diskOfferingId)
+        Map<String, Map<Integer, String>> extraDhcpOptionMap, Long diskOfferingId, Volume volume, Snapshot snapshot)
         throws InsufficientCapacityException {
 
         // VirtualMachineEntityImpl vmEntity = new VirtualMachineEntityImpl(id, owner, hostName, displayName, cpu, speed, memory, computeTags, rootDiskTags, networks, vmEntityManager);
@@ -321,7 +328,7 @@ public class CloudOrchestrator implements OrchestrationService {
 
         HypervisorType hypervisorType = HypervisorType.valueOf(hypervisor);
 
-        _itMgr.allocate(vm.getInstanceName(), _templateDao.findById(new Long(isoId)), computeOffering, rootDiskOfferingInfo, new ArrayList<DiskOfferingInfo>(), networkIpMap, plan, hypervisorType, extraDhcpOptionMap, null);
+        _itMgr.allocate(vm.getInstanceName(), _templateDao.findByIdIncludingRemoved(new Long(isoId)), computeOffering, rootDiskOfferingInfo, new ArrayList<DiskOfferingInfo>(), networkIpMap, plan, hypervisorType, extraDhcpOptionMap, null, volume, snapshot);
 
         return vmEntity;
     }

engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java

@@ -57,6 +57,8 @@ import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreDriver;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreProvider;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreProviderManager;
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStoreDriver;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
@@ -272,6 +274,9 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     @Inject
     protected SnapshotHelper snapshotHelper;
 
+    @Inject
+    private DataStoreProviderManager dataStoreProviderMgr;
+
     private final StateMachine2<Volume.State, Volume.Event, Volume> _volStateMachine;
     protected List<StoragePoolAllocator> _storagePoolAllocators;
 
@@ -898,10 +903,20 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     }
 
     private DiskProfile allocateTemplatedVolume(Type type, String name, DiskOffering offering, Long rootDisksize, Long minIops, Long maxIops, VirtualMachineTemplate template, VirtualMachine vm,
-                                                Account owner, long deviceId, String configurationId) {
+                                                Account owner, long deviceId, String configurationId, Volume volume, Snapshot snapshot) {
         assert (template.getFormat() != ImageFormat.ISO) : "ISO is not a template.";
 
-        Long size = _tmpltMgr.getTemplateSize(template, vm.getDataCenterId());
+        if (volume != null) {
+            volume = attachExistingVolumeToVm(vm, deviceId, volume, type);
+            provideVmInfoToTheStorageVolume(vm, volume);
+            return toDiskProfile(volume, offering);
+        }
+        Long size;
+        if (snapshot != null) {
+            size = _volsDao.findByIdIncludingRemoved(snapshot.getVolumeId()).getSize();
+        } else {
+            size = _tmpltMgr.getTemplateSize(template, vm.getDataCenterId());
+        }
         if (rootDisksize != null) {
             if (template.isDeployAsIs()) {
                 // Volume size specified from template deploy-as-is
@@ -961,9 +976,45 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
 
             _resourceLimitMgr.incrementVolumeResourceCount(vm.getAccountId(), vol.isDisplayVolume(), vol.getSize(), offering);
         }
+        if (snapshot != null) {
+            UserVmVO userVmVO = _userVmDao.findById(vm.getId());
+            try {
+                VolumeInfo volumeInfo = createVolumeFromSnapshot(vol, snapshot, userVmVO);
+                return toDiskProfile(volumeInfo, offering);
+            } catch (StorageUnavailableException ex) {
+                throw new CloudRuntimeException("Could not create volume from a snapshot", ex);
+            }
+        }
         return toDiskProfile(vol, offering);
     }
 
+    private void provideVmInfoToTheStorageVolume(VirtualMachine vm, Volume volume) {
+
+        StoragePoolVO pool = _storagePoolDao.findById(volume.getPoolId());
+        if (pool != null) {
+            DataStoreProvider storeProvider = dataStoreProviderMgr
+                    .getDataStoreProvider(pool.getStorageProviderName());
+            DataStoreDriver storeDriver = storeProvider.getDataStoreDriver();
+            if (storeDriver != null && storeDriver instanceof PrimaryDataStoreDriver && ((PrimaryDataStoreDriver) storeDriver).isVmInfoNeeded()) {
+                ((PrimaryDataStoreDriver) storeDriver).provideVmInfo(vm.getId(), volume.getId());
+            }
+        }
+    }
+
+    private Volume attachExistingVolumeToVm(VirtualMachine vm, long deviceId, Volume volume, Type type) {
+        VolumeVO volumeVO = _volumeDao.findById(volume.getId());
+        if (volumeVO == null) {
+            throw new CloudRuntimeException(String.format("Could not find the volume %s in the DB", volume));
+        }
+        volumeVO.setDeviceId(deviceId);
+        volumeVO.setVolumeType(type);
+        if (vm != null) {
+            volumeVO.setInstanceId(vm.getId());
+        }
+        _volumeDao.update(volumeVO.getId(), volumeVO);
+        return volumeVO;
+    }
+
     @Override
     public void saveVolumeDetails(Long diskOfferingId, Long volumeId) {
         List<VolumeDetailVO> volumeDetailsVO = new ArrayList<>();
@@ -993,7 +1044,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CREATE, eventDescription = "creating ROOT volume", create = true)
     @Override
     public List<DiskProfile> allocateTemplatedVolumes(Type type, String name, DiskOffering offering, Long rootDisksize, Long minIops, Long maxIops, VirtualMachineTemplate template, VirtualMachine vm,
-                                                      Account owner) {
+                                                      Account owner, Volume volume, Snapshot snapshot) {
         String templateToString = getReflectOnlySelectedFields(template);
 
         int volumesNumber = 1;
@@ -1040,7 +1091,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
             }
             logger.info("Adding disk object [{}] to VM [{}]", volumeName, vm);
             DiskProfile diskProfile = allocateTemplatedVolume(type, volumeName, offering, volumeSize, minIops, maxIops,
-                    template, vm, owner, deviceId, configurationId);
+                    template, vm, owner, deviceId, configurationId, volume, snapshot);
             profiles.add(diskProfile);
         }
 

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterResourceModifierActionWorker.java

@@ -424,13 +424,13 @@ public class KubernetesClusterResourceModifierActionWorker extends KubernetesClu
                     hostName, hostName, null, null, null, Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST,base64UserData, null, null, keypairs,
                     null, addrs, null, null, Objects.nonNull(affinityGroupId) ?
                             Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null,
-                    null, true, null, UserVmManager.CKS_NODE);
+                    null, true, null, UserVmManager.CKS_NODE, null, null);
         } else {
             nodeVm = userVmService.createAdvancedVirtualMachine(zone, serviceOffering, workerNodeTemplate, networkIds, owner,
                     hostName, hostName, null, null, null,
                     Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST, base64UserData, null, null, keypairs,
                     null, addrs, null, null, Objects.nonNull(affinityGroupId) ?
-                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null);
+                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null, null, null);
         }
         if (logger.isInfoEnabled()) {
             logger.info("Created node VM : {}, {} in the Kubernetes cluster : {}", hostName, nodeVm, kubernetesCluster.getName());

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/actionworkers/KubernetesClusterStartWorker.java

@@ -279,13 +279,13 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
             hostName, hostName, null, null, null, Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST,base64UserData, userDataId, userDataDetails, keypairs,
                     requestedIps, addrs, null, null, Objects.nonNull(affinityGroupId) ?
                             Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null,
-                    null, true, null, UserVmManager.CKS_NODE);
+                    null, true, null, UserVmManager.CKS_NODE, null, null);
         } else {
             controlVm = userVmService.createAdvancedVirtualMachine(zone, serviceOffering, controlNodeTemplate, networkIds, owner,
                     hostName, hostName, null, null, null,
                     Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST, base64UserData, userDataId, userDataDetails, keypairs,
                     requestedIps, addrs, null, null, Objects.nonNull(affinityGroupId) ?
-                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null);
+                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null, null, null);
         }
         if (logger.isInfoEnabled()) {
             logger.info("Created control VM: {}, {} in the Kubernetes cluster: {}", controlVm, hostName, kubernetesCluster);
@@ -447,13 +447,13 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
                     hostName, hostName, null, null, null, Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST,base64UserData, null, null, keypairs,
                     null, addrs, null, null, Objects.nonNull(affinityGroupId) ?
                             Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null,
-                    null, true, null, UserVmManager.CKS_NODE);
+                    null, true, null, UserVmManager.CKS_NODE, null, null);
         } else {
             additionalControlVm = userVmService.createAdvancedVirtualMachine(zone, serviceOffering, controlNodeTemplate, networkIds, owner,
                     hostName, hostName, null, null, null,
                     Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST, base64UserData, null, null, keypairs,
                     null, addrs, null, null, Objects.nonNull(affinityGroupId) ?
-                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null);
+                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null, null, null);
         }
 
         if (logger.isInfoEnabled()) {
@@ -491,13 +491,13 @@ public class KubernetesClusterStartWorker extends KubernetesClusterResourceModif
                     hostName, hostName, null, null, null, Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST,base64UserData, null, null, keypairs,
                     Map.of(kubernetesCluster.getNetworkId(), requestedIps.get(etcdNodeIndex)), addrs, null, null, Objects.nonNull(affinityGroupId) ?
                             Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null,
-                    null, true, null, null);
+                    null, true, null, null, null, null);
         } else {
             etcdNode = userVmService.createAdvancedVirtualMachine(zone, serviceOffering, etcdTemplate, networkIds, owner,
                     hostName, hostName, null, null, null,
                     Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST, base64UserData, null, null, keypairs,
                     Map.of(kubernetesCluster.getNetworkId(), requestedIps.get(etcdNodeIndex)), addrs, null, null, Objects.nonNull(affinityGroupId) ?
-                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null);
+                            Collections.singletonList(affinityGroupId) : null, customParameterMap, null, null, null, null, true, UserVmManager.CKS_NODE, null, null, null);
         }
 
         if (logger.isInfoEnabled()) {

plugins/network-elements/elastic-loadbalancer/src/main/java/com/cloud/network/lb/LoadBalanceRuleHandler.java

@@ -297,7 +297,7 @@ public class LoadBalanceRuleHandler {
                         elasticLbVmOffering.isOfferHA(), false, null);
                 elbVm.setRole(Role.LB);
                 elbVm = _routerDao.persist(elbVm);
-                _itMgr.allocate(elbVm.getInstanceName(), template, elasticLbVmOffering, networks, plan, null);
+                _itMgr.allocate(elbVm.getInstanceName(), template, elasticLbVmOffering, networks, plan, null, null, null);
                 elbVm = _routerDao.findById(elbVm.getId());
                 //TODO: create usage stats
             }

plugins/network-elements/internal-loadbalancer/src/main/java/org/apache/cloudstack/network/lb/InternalLoadBalancerVMManagerImpl.java

@@ -780,7 +780,7 @@ public class InternalLoadBalancerVMManagerImpl extends ManagerBase implements In
             try {
                 internalLbVm = createOrUpdateInternalLb(internalLbVm, id, internalLbProviderId, owner, userId, vpcId,
                         routerOffering, template);
-                _itMgr.allocate(internalLbVm.getInstanceName(), template, routerOffering, networks, plan, null);
+                _itMgr.allocate(internalLbVm.getInstanceName(), template, routerOffering, networks, plan, null, null, null);
                 internalLbVm = _internalLbVmDao.findById(internalLbVm.getId());
                 if (templatesIterator.hasNext()) {
                     _itMgr.checkDeploymentPlan(internalLbVm, template, routerOffering, owner, plan);

plugins/network-elements/internal-loadbalancer/src/test/java/org/apache/cloudstack/network/lb/InternalLoadBalancerVMManagerImplTest.java

@@ -134,7 +134,7 @@ public class InternalLoadBalancerVMManagerImplTest {
         when(internalLbVmDao.findById(anyLong())).thenReturn(mock(DomainRouterVO.class));
         DomainRouterVO result = service.deployInternalLbVmWithTemplates(null, id, plan, internalLbProviderId, account, userId, vpcId, serviceOffering, networks, templates);
         assertNotNull(result);
-        verify(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull());
+        verify(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull(), isNull(), isNull());
     }
 
     @Test
@@ -149,11 +149,11 @@ public class InternalLoadBalancerVMManagerImplTest {
         when(internalLbVmDao.persist(any(DomainRouterVO.class))).thenAnswer(invocation -> invocation.getArgument(0));
         when(internalLbVmDao.findById(anyLong())).thenReturn(mock(DomainRouterVO.class));
         doThrow(new InsufficientServerCapacityException("Not enough capacity", id))
-                .when(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull());
+                .when(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull(), isNull(), isNull());
         DomainRouterVO result = service.deployInternalLbVmWithTemplates(null, id, plan, internalLbProviderId, account, userId, vpcId, serviceOffering, networks, templates);
         assertNotNull(result);
-        verify(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull());
+        verify(virtualMachineManager).allocate(anyString(), eq(template1), eq(serviceOffering), eq(networks), eq(plan), isNull(), isNull(), isNull());
-        verify(virtualMachineManager).allocate(anyString(), eq(template2), eq(serviceOffering), eq(networks), eq(plan), isNull());
+        verify(virtualMachineManager).allocate(anyString(), eq(template2), eq(serviceOffering), eq(networks), eq(plan), isNull(), isNull(), isNull());
     }
 
     @Test(expected = InsufficientCapacityException.class)
@@ -166,7 +166,7 @@ public class InternalLoadBalancerVMManagerImplTest {
         LinkedHashMap<Network, List<? extends NicProfile>> networks = new LinkedHashMap<>();
         when(internalLbVmDao.persist(any(DomainRouterVO.class))).thenAnswer(invocation -> invocation.getArgument(0));
         doThrow(new InsufficientServerCapacityException("Insufficient capacity", id))
-                .when(virtualMachineManager).allocate(anyString(), any(VMTemplateVO.class), eq(serviceOffering), eq(networks), eq(plan), isNull());
+                .when(virtualMachineManager).allocate(anyString(), any(VMTemplateVO.class), eq(serviceOffering), eq(networks), eq(plan), isNull(), isNull(), isNull());
         service.deployInternalLbVmWithTemplates(null, id, plan, internalLbProviderId, account, userId, vpcId, serviceOffering, networks, templates);
     }
 }

plugins/network-elements/juniper-contrail/src/main/java/org/apache/cloudstack/network/contrail/management/ServiceManagerImpl.java

@@ -130,7 +130,7 @@ public class ServiceManagerImpl implements ServiceManager {
         svm.setUserData(userData);
 
         try {
-            _vmManager.allocate(instanceName, template, serviceOffering, networks, plan, template.getHypervisorType());
+            _vmManager.allocate(instanceName, template, serviceOffering, networks, plan, template.getHypervisorType(), null, null);
         } catch (InsufficientCapacityException ex) {
             throw new CloudRuntimeException("Insufficient capacity", ex);
         }

plugins/network-elements/netscaler/src/main/java/com/cloud/network/vm/NetScalerVMManagerImpl.java

@@ -356,7 +356,7 @@ public class NetScalerVMManagerImpl extends ManagerBase implements NetScalerVMMa
         nsVpx = _routerDao.persist(nsVpx);
 
         VMInstanceVO vmVO= _vmDao.findVMByHostName(nxVpxName);
-        _itMgr.allocate(nxVpxName, template, vpxOffering, networks, plan, template.getHypervisorType());
+        _itMgr.allocate(nxVpxName, template, vpxOffering, networks, plan, template.getHypervisorType(), null, null);
         Map<Param, Object> params = new HashMap<VirtualMachineProfile.Param, Object>(1);
         try {
             if (vmVO != null) {

plugins/storage/sharedfs/storagevm/src/main/java/org/apache/cloudstack/storage/sharedfs/lifecycle/StorageVmSharedFSLifeCycle.java

@@ -17,29 +17,41 @@
 
 package org.apache.cloudstack.storage.sharedfs.lifecycle;
 
-import static org.apache.cloudstack.storage.sharedfs.SharedFS.SharedFSPath;
+import com.cloud.dc.DataCenter;
-import static org.apache.cloudstack.storage.sharedfs.SharedFS.SharedFSVmNamePrefix;
-import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_CPU_COUNT;
-import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_RAM_SIZE;
-
 import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.exception.InsufficientCapacityException;
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.exception.ManagementServerException;
 import com.cloud.exception.OperationTimedoutException;
 import com.cloud.exception.ResourceAllocationException;
+import com.cloud.exception.ResourceUnavailableException;
 import com.cloud.exception.VirtualMachineMigrationException;
+import com.cloud.hypervisor.Hypervisor;
+import com.cloud.network.Network;
 import com.cloud.offering.ServiceOffering;
+import com.cloud.resource.ResourceManager;
+import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.LaunchPermissionVO;
+import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeApiService;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.LaunchPermissionDao;
+import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VolumeDao;
+import com.cloud.user.Account;
+import com.cloud.user.AccountManager;
 import com.cloud.uservm.UserVm;
 import com.cloud.utils.FileUtil;
 import com.cloud.utils.Pair;
-
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.utils.net.NetUtils;
+import com.cloud.vm.UserVmManager;
+import com.cloud.vm.UserVmService;
+import com.cloud.vm.UserVmVO;
+import com.cloud.vm.VirtualMachineManager;
+import com.cloud.vm.dao.NicDao;
+import com.cloud.vm.dao.UserVmDao;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -47,17 +59,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-
 import javax.inject.Inject;
-
-import com.cloud.exception.ResourceUnavailableException;
-import com.cloud.utils.net.NetUtils;
-import com.cloud.vm.UserVmManager;
-import com.cloud.vm.UserVmService;
-import com.cloud.vm.UserVmVO;
-import com.cloud.vm.dao.NicDao;
-import com.cloud.vm.dao.UserVmDao;
-
 import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.api.BaseCmd;
 import org.apache.cloudstack.context.CallContext;
@@ -67,17 +69,11 @@ import org.apache.commons.codec.binary.Base64;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-import com.cloud.dc.DataCenter;
+
-import com.cloud.hypervisor.Hypervisor;
+import static org.apache.cloudstack.storage.sharedfs.SharedFS.SharedFSPath;
-import com.cloud.network.Network;
+import static org.apache.cloudstack.storage.sharedfs.SharedFS.SharedFSVmNamePrefix;
-import com.cloud.resource.ResourceManager;
+import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_CPU_COUNT;
-import com.cloud.service.dao.ServiceOfferingDao;
+import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_RAM_SIZE;
-import com.cloud.storage.VMTemplateVO;
-import com.cloud.storage.dao.VMTemplateDao;
-import com.cloud.user.Account;
-import com.cloud.user.AccountManager;
-import com.cloud.utils.exception.CloudRuntimeException;
-import com.cloud.vm.VirtualMachineManager;
 
 public class StorageVmSharedFSLifeCycle implements SharedFSLifeCycle {
     protected Logger logger = LogManager.getLogger(getClass());
@@ -197,7 +193,7 @@ public class StorageVmSharedFSLifeCycle implements SharedFSLifeCycle {
                         diskOfferingId, size, null, Hypervisor.HypervisorType.None, BaseCmd.HTTPMethod.POST, base64UserData,
                         null, null, keypairs, null, addrs, null, null, null,
                         customParameterMap, null, null, null, null,
-                        true, UserVmManager.SHAREDFSVM, null);
+                        true, UserVmManager.SHAREDFSVM, null, null, null);
                 vmContext.setEventResourceId(vm.getId());
                 userVmService.startVirtualMachine(vm, null);
             } catch (InsufficientCapacityException ex) {

plugins/storage/sharedfs/storagevm/src/test/java/org/apache/cloudstack/storage/sharedfs/lifecycle/StorageVmSharedFSLifeCycleTest.java

@@ -17,39 +17,6 @@
 
 package org.apache.cloudstack.storage.sharedfs.lifecycle;
 
-import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_CPU_COUNT;
-import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_RAM_SIZE;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyMap;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.isNull;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockStatic;
-import static org.mockito.Mockito.when;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Optional;
-
-import org.apache.cloudstack.api.ApiCommandResourceType;
-import org.apache.cloudstack.api.BaseCmd;
-import org.apache.cloudstack.context.CallContext;
-import org.apache.cloudstack.storage.sharedfs.SharedFS;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.MockedStatic;
-import org.mockito.MockitoAnnotations;
-import org.mockito.Spy;
-import org.mockito.junit.MockitoJUnitRunner;
-
 import com.cloud.dc.DataCenter;
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.dao.DataCenterDao;
@@ -85,6 +52,38 @@ import com.cloud.vm.UserVmVO;
 import com.cloud.vm.VirtualMachineManager;
 import com.cloud.vm.dao.NicDao;
 import com.cloud.vm.dao.UserVmDao;
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import org.apache.cloudstack.api.ApiCommandResourceType;
+import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.storage.sharedfs.SharedFS;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+
+import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_CPU_COUNT;
+import static org.apache.cloudstack.storage.sharedfs.provider.StorageVmSharedFSProvider.SHAREDFSVM_MIN_RAM_SIZE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
 
 @RunWith(MockitoJUnitRunner.class)
 public class StorageVmSharedFSLifeCycleTest {
@@ -258,7 +257,7 @@ public class StorageVmSharedFSLifeCycleTest {
                 anyString(), anyLong(), anyLong(), isNull(), any(Hypervisor.HypervisorType.class), any(BaseCmd.HTTPMethod.class), anyString(),
                 isNull(), isNull(), anyList(), isNull(), any(Network.IpAddresses.class), isNull(), isNull(), isNull(),
                 anyMap(), isNull(), isNull(), isNull(), isNull(),
-                anyBoolean(), anyString(), isNull())).thenReturn(vm);
+                anyBoolean(), anyString(), isNull(), isNull(), isNull())).thenReturn(vm);
 
         VolumeVO volume = mock(VolumeVO.class);
         when(volume.getId()).thenReturn(s_volumeId);

plugins/storage/volume/storpool/src/main/java/org/apache/cloudstack/storage/datastore/driver/StorPoolPrimaryDataStoreDriver.java

@@ -37,6 +37,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreState
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStoreDriver;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
 import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
@@ -171,6 +172,8 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
     private ServiceOfferingDetailsDao serviceOfferingDetailDao;
     @Inject
     private ServiceOfferingDao serviceOfferingDao;
+    @Inject
+    private VolumeDataFactory volumeDataFactory;
 
     private SnapshotDataStoreVO getSnapshotImageStoreRef(long snapshotId, long zoneId) {
         List<SnapshotDataStoreVO> snaps = snapshotDataStoreDao.listReadyBySnapshot(snapshotId, DataStoreRole.Image);
@@ -626,8 +629,11 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
                 VolumeInfo vinfo = (VolumeInfo)dstData;
                 final String volumeName = vinfo.getUuid();
                 final Long size = vinfo.getSize();
+
                 SpConnectionDesc conn = StorPoolUtil.getSpConnection(vinfo.getDataStore().getUuid(), vinfo.getDataStore().getId(), storagePoolDetailsDao, primaryStoreDao);
-                SpApiResponse resp = StorPoolUtil.volumeCreate(volumeName, snapshotName, size, null, null, "volume", sinfo.getBaseVolume().getMaxIops(), conn);
+
+                StorPoolVolumeDef spVolume = createVolumeWithTags(sinfo, snapshotName, vinfo, volumeName, size, conn);
+                SpApiResponse resp = StorPoolUtil.volumeCreate(spVolume, conn);
                 if (resp.getError() == null) {
                     updateStoragePool(dstData.getDataStore().getId(), size);
 
@@ -643,9 +649,10 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
                     SnapshotDataStoreVO snap = getSnapshotImageStoreRef(sinfo.getId(), vinfo.getDataCenterId());
                     SnapshotDetailsVO snapshotDetail = snapshotDetailsDao.findDetail(sinfo.getId(), StorPoolUtil.SP_DELAY_DELETE);
                     if (snapshotDetail != null) {
-                        err = String.format("Could not create volume from snapshot due to: %s. The snapshot was created with the delayDelete option.", resp.getError());
+                        answer = new Answer(cmd, false, String.format("Could not create volume from snapshot due to: %s. The snapshot was created with the delayDelete option.", resp.getError()));
                     } else if (snap != null && StorPoolStorageAdaptor.getVolumeNameFromPath(snap.getInstallPath(), false) == null) {
-                        SpApiResponse emptyVolumeCreateResp = StorPoolUtil.volumeCreate(volumeName, null, size, null, null, "volume", null, conn);
+                        spVolume.setParent(null);
+                        SpApiResponse emptyVolumeCreateResp = StorPoolUtil.volumeCreate(spVolume, conn);
                         if (emptyVolumeCreateResp.getError() == null) {
                             answer = createVolumeFromSnapshot(srcData, dstData, size, emptyVolumeCreateResp);
                         } else {
@@ -655,7 +662,7 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
                         answer = new Answer(cmd, false, String.format("The snapshot %s does not exists neither on primary, neither on secondary storage. Cannot create volume from snapshot", snapshotName));
                     }
                 } else {
-                    err = String.format("Could not create Storpool volume %s from snapshot %s. Error: %s", volumeName, snapshotName, resp.getError());
+                    answer = new Answer(cmd, false, String.format("Could not create Storpool volume %s from snapshot %s. Error: %s", volumeName, snapshotName, resp.getError()));
                 }
             } else if (srcType == DataObjectType.SNAPSHOT && dstType == DataObjectType.SNAPSHOT) {
                 SnapshotInfo sinfo = (SnapshotInfo)srcData;
@@ -982,6 +989,27 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
         callback.complete(res);
     }
 
+    private StorPoolVolumeDef createVolumeWithTags(SnapshotInfo sinfo, String snapshotName, VolumeInfo vinfo, String volumeName, Long size, SpConnectionDesc conn) {
+        Pair<String, String> templateAndTier = getTemplateAndTier(vinfo, conn);
+        Map<String, String> tags = StorPoolHelper.addStorPoolTags(volumeName, getVMInstanceUUID(vinfo.getInstanceId()), "volume", getVcPolicyTag(vinfo.getInstanceId()), templateAndTier.first());
+        return new StorPoolVolumeDef(null, size, tags, snapshotName, sinfo.getBaseVolume().getMaxIops(), templateAndTier.second(), null, null, null);
+    }
+
+    private Pair<String, String> getTemplateAndTier(VolumeInfo vinfo, SpConnectionDesc conn) {
+        String tier = null;
+        String template = null;
+        if (vinfo.getDiskOfferingId() != null) {
+            tier = getTierFromOfferingDetail(vinfo.getDiskOfferingId());
+            if (tier == null) {
+                template = getTemplateFromOfferingDetail(vinfo.getDiskOfferingId());
+            }
+        }
+
+        if (template == null) {
+            template = conn.getTemplateName();
+        }
+        return new Pair<>(tier, template);
+    }
     private Answer createVolumeSnapshot(StorageSubSystemCommand cmd, Long size, SpConnectionDesc conn,
             String volName, TemplateObjectTO dstTO) {
         Answer answer;
@@ -1302,24 +1330,33 @@ public class StorPoolPrimaryDataStoreDriver implements PrimaryDataStoreDriver {
             return;
         }
         StoragePoolVO poolVO = primaryStoreDao.findById(volume.getPoolId());
-        if (poolVO != null) {
+        if (poolVO != null && StoragePoolType.StorPool.equals(poolVO.getPoolType())) {
-            try {
+            VolumeInfo vInfo = volumeDataFactory.getVolume(volumeId);
-                SpConnectionDesc conn = StorPoolUtil.getSpConnection(poolVO.getUuid(), poolVO.getId(), storagePoolDetailsDao, primaryStoreDao);
+            if (vInfo == null) {
-                String volName = StorPoolStorageAdaptor.getVolumeNameFromPath(volume.getPath(), true);
+                StorPoolUtil.spLog("Could not find volume with volume ID [%s] to set tags", volumeId);
-                VMInstanceVO userVM = vmInstanceDao.findById(vmId);
+                return;
-                Map<String, String> tags = StorPoolHelper.addStorPoolTags(null, userVM.getUuid(), null, getVcPolicyTag(vmId), null);
-                if (volume.getDeviceId() != null) {
-                    tags.put("disk", volume.getDeviceId().toString());
-                }
-                StorPoolVolumeDef spVolume = new StorPoolVolumeDef(volName, null, tags, null, null, null, null, null, null);
-
-                SpApiResponse resp = StorPoolUtil.volumeUpdate(spVolume, conn);
-                if (resp.getError() != null) {
-                    logger.warn(String.format("Could not update VC policy tags of a volume with id [%s]", volume.getUuid()));
-                }
-            } catch (Exception e) {
-                logger.warn(String.format("Could not update Virtual machine tags due to %s", e.getMessage()));
             }
+            updateVolumeWithTags(poolVO, vInfo);
+        }
+    }
+
+    private void updateVolumeWithTags(StoragePoolVO poolVO, VolumeInfo vInfo) {
+        try {
+            SpConnectionDesc conn = StorPoolUtil.getSpConnection(poolVO.getUuid(), poolVO.getId(), storagePoolDetailsDao, primaryStoreDao);
+            String volName = StorPoolStorageAdaptor.getVolumeNameFromPath(vInfo.getPath(), true);
+            Pair<String, String> templateAndTier = getTemplateAndTier(vInfo, conn);
+            Map<String, String> tags = StorPoolHelper.addStorPoolTags(volName, getVMInstanceUUID(vInfo.getInstanceId()), "volume", getVcPolicyTag(vInfo.getInstanceId()), templateAndTier.first());
+            if (vInfo.getDeviceId() != null) {
+                tags.put("disk", vInfo.getDeviceId().toString());
+            }
+            StorPoolVolumeDef spVolume = new StorPoolVolumeDef(volName, null, tags, null, null, templateAndTier.second(), null, null, null);
+            StorPoolUtil.spLog("Updating volume's tags [%s] with template [%s]", tags, templateAndTier.second());
+            SpApiResponse resp = StorPoolUtil.volumeUpdate(spVolume, conn);
+            if (resp.getError() != null) {
+                logger.warn(String.format("Could not update VC policy tags of a volume with id [%s]", vInfo.getUuid()));
+            }
+        } catch (Exception e) {
+            logger.warn(String.format("Could not update Virtual machine tags due to %s", e.getMessage()));
         }
     }
 

server/src/main/java/com/cloud/consoleproxy/ConsoleProxyManagerImpl.java

@@ -725,7 +725,7 @@ public class ConsoleProxyManagerImpl extends ManagerBase implements ConsoleProxy
             proxy = createOrUpdateConsoleProxy(proxy, dataCenterId, id, name, serviceOffering, template, systemAcct);
             try {
                 virtualMachineManager.allocate(name, template, serviceOffering, networks, plan,
-                        template.getHypervisorType());
+                        template.getHypervisorType(), null, null);
                 proxy = consoleProxyDao.findById(proxy.getId());
                 virtualMachineManager.checkDeploymentPlan(proxy, template, serviceOffering, systemAcct, plan);
                 break;

server/src/main/java/com/cloud/network/as/AutoScaleManagerImpl.java

@@ -1810,7 +1810,7 @@ public class AutoScaleManagerImpl extends ManagerBase implements AutoScaleManage
                         vmHostName, diskOfferingId, dataDiskSize, null,
                         hypervisorType, HTTPMethod.GET, userData, userDataId, userDataDetails, sshKeyPairs,
                         null, null, true, null, affinityGroupIdList, customParameters, null, null, null,
-                        null, true, overrideDiskOfferingId);
+                        null, true, overrideDiskOfferingId, null, null);
             } else {
                 if (networkModel.checkSecurityGroupSupportForNetwork(owner, zone, networkIds,
                         Collections.emptyList())) {
@@ -1818,13 +1818,13 @@ public class AutoScaleManagerImpl extends ManagerBase implements AutoScaleManage
                             owner, vmHostName,vmHostName, diskOfferingId, dataDiskSize, null,
                             hypervisorType, HTTPMethod.GET, userData, userDataId, userDataDetails, sshKeyPairs,
                             null, null, true, null, affinityGroupIdList, customParameters, null, null, null,
-                            null, true, overrideDiskOfferingId, null);
+                            null, true, overrideDiskOfferingId, null, null, null);
                 } else {
                     vm = userVmService.createAdvancedVirtualMachine(zone, serviceOffering, template, networkIds, owner, vmHostName, vmHostName,
                             diskOfferingId, dataDiskSize, null,
                             hypervisorType, HTTPMethod.GET, userData, userDataId, userDataDetails, sshKeyPairs,
                             null, addrs, true, null, affinityGroupIdList, customParameters, null, null, null,
-                            null, true, null, overrideDiskOfferingId);
+                            null, true, null, overrideDiskOfferingId, null, null);
                 }
             }
 

server/src/main/java/com/cloud/network/router/NetworkHelperImpl.java

@@ -889,7 +889,7 @@ public class NetworkHelperImpl implements NetworkHelper {
 
         final LinkedHashMap<Network, List<? extends NicProfile>> networks = configureDefaultNics(routerDeploymentDefinition);
 
-        _itMgr.allocate(router.getInstanceName(), template, routerOffering, networks, routerDeploymentDefinition.getPlan(), hType);
+        _itMgr.allocate(router.getInstanceName(), template, routerOffering, networks, routerDeploymentDefinition.getPlan(), hType, null, null);
     }
 
     public static void setSystemAccount(final Account systemAccount) {

server/src/main/java/com/cloud/network/router/VpcNetworkHelperImpl.java

@@ -169,7 +169,7 @@ public class VpcNetworkHelperImpl extends NetworkHelperImpl {
             }
         }
 
-        _itMgr.allocate(router.getInstanceName(), template, routerOffering, networks, vpcRouterDeploymentDefinition.getPlan(), hType);
+        _itMgr.allocate(router.getInstanceName(), template, routerOffering, networks, vpcRouterDeploymentDefinition.getPlan(), hType, null, null);
     }
 
     @Override

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

@@ -397,6 +397,10 @@ import com.cloud.vm.snapshot.VMSnapshotManager;
 import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
+
+
 public class UserVmManagerImpl extends ManagerBase implements UserVmManager, VirtualMachineGuru, Configurable {
 
     /**
@@ -617,6 +621,8 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     @Inject
     NetworkService networkService;
 
+    @Inject
+    SnapshotDataFactory snapshotDataFactory;
 
     private ScheduledExecutorService _executor = null;
     private ScheduledExecutorService _vmIpFetchExecutor = null;
@@ -3705,7 +3711,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                                          Account owner, String hostName, String displayName, Long diskOfferingId, Long diskSize, String group, HypervisorType hypervisor, HTTPMethod httpmethod,
                                                          String userData, Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean displayVm, String keyboard, List<Long> affinityGroupIdList,
                                                          Map<String, String> customParametes, String customId, Map<String, Map<Integer, String>> dhcpOptionMap,
-                                                         Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException,
+                                                         Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException,
     StorageUnavailableException, ResourceAllocationException {
 
         Account caller = CallContext.current().getCallingAccount();
@@ -3754,7 +3760,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
         return createVirtualMachine(zone, serviceOffering, template, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod,
                 userData, userDataId, userDataDetails, sshKeyPairs, hypervisor, caller, requestedIps, defaultIps, displayVm, keyboard, affinityGroupIdList, customParametes, customId, dhcpOptionMap,
-                dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, null, overrideDiskOfferingId);
+                dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, null, overrideDiskOfferingId, volume, snapshot);
 
     }
 
@@ -3764,7 +3770,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                                             List<Long> securityGroupIdList, Account owner, String hostName, String displayName, Long diskOfferingId, Long diskSize, String group, HypervisorType hypervisor,
                                                             HTTPMethod httpmethod, String userData, Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean displayVm, String keyboard,
                                                             List<Long> affinityGroupIdList, Map<String, String> customParameters, String customId, Map<String, Map<Integer, String>> dhcpOptionMap,
-                                                            Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, String vmType) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException {
+                                                            Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap, Map<String, String> userVmOVFProperties, boolean dynamicScalingEnabled, Long overrideDiskOfferingId, String vmType, Volume volume, Snapshot snapshot) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException, StorageUnavailableException, ResourceAllocationException {
 
         Account caller = CallContext.current().getCallingAccount();
         List<NetworkVO> networkList = new ArrayList<NetworkVO>();
@@ -3867,7 +3873,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
         return createVirtualMachine(zone, serviceOffering, template, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod,
                 userData, userDataId, userDataDetails, sshKeyPairs, hypervisor, caller, requestedIps, defaultIps, displayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, dataDiskTemplateToDiskOfferingMap,
-                userVmOVFProperties, dynamicScalingEnabled, vmType, overrideDiskOfferingId);
+                userVmOVFProperties, dynamicScalingEnabled, vmType, overrideDiskOfferingId, volume, snapshot);
     }
 
     @Override
@@ -3876,7 +3882,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                                String hostName, String displayName, Long diskOfferingId, Long diskSize, String group, HypervisorType hypervisor, HTTPMethod httpmethod, String userData,
                                                Long userDataId, String userDataDetails, List<String> sshKeyPairs, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean displayvm, String keyboard, List<Long> affinityGroupIdList,
                                                Map<String, String> customParametrs, String customId, Map<String, Map<Integer, String>> dhcpOptionsMap, Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                               Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException,
+                                               Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException, ConcurrentOperationException, ResourceUnavailableException,
     StorageUnavailableException, ResourceAllocationException {
 
         Account caller = CallContext.current().getCallingAccount();
@@ -3929,7 +3935,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         verifyExtraDhcpOptionsNetwork(dhcpOptionsMap, networkList);
         return createVirtualMachine(zone, serviceOffering, template, hostName, displayName, owner, diskOfferingId, diskSize, networkList, null, group, httpmethod, userData,
                 userDataId, userDataDetails, sshKeyPairs, hypervisor, caller, requestedIps, defaultIps, displayvm, keyboard, affinityGroupIdList, customParametrs, customId, dhcpOptionsMap,
-                dataDiskTemplateToDiskOfferingMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, overrideDiskOfferingId);
+                dataDiskTemplateToDiskOfferingMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, overrideDiskOfferingId, volume, snapshot);
     }
 
     @Override
@@ -4061,7 +4067,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                         Long userDataId, String userDataDetails, List<String> sshKeyPairs, HypervisorType hypervisor, Account caller, Map<Long, IpAddresses> requestedIps, IpAddresses defaultIps, Boolean isDisplayVm, String keyboard,
                                         List<Long> affinityGroupIdList, Map<String, String> customParameters, String customId, Map<String, Map<Integer, String>> dhcpOptionMap,
                                         Map<Long, DiskOffering> datadiskTemplateToDiskOfferringMap,
-                                        Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId) throws InsufficientCapacityException, ResourceUnavailableException,
+                                        Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, Long overrideDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException, ResourceUnavailableException,
     ConcurrentOperationException, StorageUnavailableException, ResourceAllocationException {
 
         _accountMgr.checkAccess(caller, null, true, owner);
@@ -4069,7 +4075,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         if (owner.getState() == Account.State.DISABLED) {
             throw new PermissionDeniedException("The owner of vm to deploy is disabled: " + owner);
         }
-        VMTemplateVO template = _templateDao.findById(tmplt.getId());
+        VMTemplateVO template = _templateDao.findByIdIncludingRemoved(tmplt.getId());
         if (template != null) {
             _templateDao.loadDetails(template);
         }
@@ -4141,10 +4147,18 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             rootDiskOfferingId = overrideDiskOfferingId;
         }
 
-        DiskOfferingVO rootdiskOffering = _diskOfferingDao.findById(rootDiskOfferingId);
+        DiskOfferingVO rootDiskOffering = _diskOfferingDao.findById(rootDiskOfferingId);
-        long volumesSize = configureCustomRootDiskSize(customParameters, template, hypervisorType, rootdiskOffering);
+        long volumesSize = 0;
+        if (volume != null) {
+            volumesSize = volume.getSize();
+        } else if (snapshot != null) {
+            VolumeVO volumeVO = _volsDao.findById(snapshot.getVolumeId());
+            volumesSize = volumeVO != null ? volumeVO.getSize() : 0;
+        } else {
+            volumesSize = configureCustomRootDiskSize(customParameters, template, hypervisorType, rootDiskOffering);
+        }
 
-        if (rootdiskOffering.getEncrypt() && hypervisorType != HypervisorType.KVM) {
+        if (rootDiskOffering.getEncrypt() && hypervisorType != HypervisorType.KVM) {
             throw new InvalidParameterValueException("Root volume encryption is not supported for hypervisor type " + hypervisorType);
         }
 
@@ -4153,7 +4167,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             DiskOfferingVO diskOffering = _diskOfferingDao.findById(diskOfferingId);
             additionalDiskSize = verifyAndGetDiskSize(diskOffering, diskSize);
         }
-        UserVm vm = getCheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize);
+        UserVm vm = getCheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize, volume, snapshot);
 
         _securityGroupMgr.addInstanceToGroups(vm, securityGroupIdList);
 
@@ -4173,14 +4187,14 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         Map<String, Map<Integer, String>> dhcpOptionMap, Map<Long, DiskOffering> datadiskTemplateToDiskOfferringMap,
         Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, VMTemplateVO template,
         HypervisorType hypervisorType, long accountId, ServiceOfferingVO offering, boolean isIso,
-        Long rootDiskOfferingId, long volumesSize, long additionalDiskSize) throws ResourceAllocationException {
+        Long rootDiskOfferingId, long volumesSize, long additionalDiskSize, Volume volume, Snapshot snapshot) throws ResourceAllocationException {
         if (!VirtualMachineManager.ResourceCountRunningVMsonly.value()) {
             List<String> resourceLimitHostTags = resourceLimitService.getResourceLimitHostTags(offering, template);
             try (CheckedReservation vmReservation = new CheckedReservation(owner, ResourceType.user_vm, resourceLimitHostTags, 1l, reservationDao, resourceLimitService);
                  CheckedReservation cpuReservation = new CheckedReservation(owner, ResourceType.cpu, resourceLimitHostTags, Long.valueOf(offering.getCpu()), reservationDao, resourceLimitService);
                  CheckedReservation memReservation = new CheckedReservation(owner, ResourceType.memory, resourceLimitHostTags, Long.valueOf(offering.getRamSize()), reservationDao, resourceLimitService);
             ) {
-                return getUncheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize);
+                return getUncheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize, volume, snapshot);
             } catch (ResourceAllocationException | CloudRuntimeException  e) {
                 throw e;
             } catch (Exception e) {
@@ -4189,7 +4203,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             }
 
         } else {
-            return getUncheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize);
+            return getUncheckedUserVmResource(zone, hostName, displayName, owner, diskOfferingId, diskSize, networkList, securityGroupIdList, group, httpmethod, userData, userDataId, userDataDetails, sshKeyPairs, caller, requestedIps, defaultIps, isDisplayVm, keyboard, affinityGroupIdList, customParameters, customId, dhcpOptionMap, datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, template, hypervisorType, accountId, offering, isIso, rootDiskOfferingId, volumesSize, additionalDiskSize, volume, snapshot);
         }
     }
 
@@ -4206,7 +4220,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         Map<String, Map<Integer, String>> dhcpOptionMap, Map<Long, DiskOffering> datadiskTemplateToDiskOfferringMap,
         Map<String, String> userVmOVFPropertiesMap, boolean dynamicScalingEnabled, String vmType, VMTemplateVO template,
         HypervisorType hypervisorType, long accountId, ServiceOfferingVO offering, boolean isIso,
-        Long rootDiskOfferingId, long volumesSize, long additionalDiskSize) throws ResourceAllocationException
+        Long rootDiskOfferingId, long volumesSize, long additionalDiskSize, Volume volume, Snapshot snapshot) throws ResourceAllocationException
     {
         List<String> rootResourceLimitStorageTags = getResourceLimitStorageTags(rootDiskOfferingId != null ? rootDiskOfferingId : offering.getDiskOfferingId());
         List<String> additionalResourceLimitStorageTags = diskOfferingId != null ? getResourceLimitStorageTags(diskOfferingId) : null;
@@ -4304,9 +4318,21 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             if (template.getTemplateType().equals(TemplateType.SYSTEM) && !CKS_NODE.equals(vmType) && !SHAREDFSVM.equals(vmType)) {
                 throw new InvalidParameterValueException(String.format("Unable to use system template %s to deploy a user vm", template));
             }
-            List<VMTemplateZoneVO> listZoneTemplate = _templateZoneDao.listByZoneTemplate(zone.getId(), template.getId());
+
-            if (listZoneTemplate == null || listZoneTemplate.isEmpty()) {
+            if (volume != null) {
-                throw new InvalidParameterValueException(String.format("The template %s is not available for use", template));
+                if (zone.getId() != volume.getDataCenterId()) {
+                    throw new InvalidParameterValueException(String.format("The volume's zone [%s] is not the same as the provided zone [%s]", volume.getDataCenterId(), zone.getId()));
+                }
+            } else if (snapshot != null) {
+                List<SnapshotInfo> snapshotsOnZone = snapshotDataFactory.getSnapshots(snapshot.getId(), zone.getId());
+                if (CollectionUtils.isEmpty(snapshotsOnZone)) {
+                    throw new InvalidParameterValueException("The snapshot does not exist on zone " + zone.getId());
+                }
+            } else {
+                List<VMTemplateZoneVO> listZoneTemplate = _templateZoneDao.listByZoneTemplate(zone.getId(), template.getId());
+                if (listZoneTemplate == null || listZoneTemplate.isEmpty()) {
+                    throw new InvalidParameterValueException("The template " + template.getId() + " is not available for use");
+                }
             }
 
             if (isIso && !template.isBootable()) {
@@ -4485,7 +4511,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
             UserVmVO vm = commitUserVm(zone, template, hostName, displayName, owner, diskOfferingId, diskSize, userData, userDataId, userDataDetails, caller, isDisplayVm, keyboard, accountId, userId, offering,
                     isIso, sshPublicKeys, networkNicMap, id, instanceName, uuidName, hypervisorType, customParameters, dhcpOptionMap,
-                    datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, rootDiskOfferingId, keypairnames);
+                    datadiskTemplateToDiskOfferringMap, userVmOVFPropertiesMap, dynamicScalingEnabled, vmType, rootDiskOfferingId, keypairnames, volume, snapshot);
 
             assignInstanceToGroup(group, id);
             return vm;
@@ -4626,7 +4652,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                   final long accountId, final long userId, final ServiceOffering offering, final boolean isIso, final String sshPublicKeys, final LinkedHashMap<String, List<NicProfile>> networkNicMap,
                                   final long id, final String instanceName, final String uuidName, final HypervisorType hypervisorType, final Map<String, String> customParameters,
                                   final Map<String, Map<Integer, String>> extraDhcpOptionMap, final Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                  final Map<String, String> userVmOVFPropertiesMap, final VirtualMachine.PowerState powerState, final boolean dynamicScalingEnabled, String vmType, final Long rootDiskOfferingId, String sshkeypairs) throws InsufficientCapacityException {
+                                  final Map<String, String> userVmOVFPropertiesMap, final VirtualMachine.PowerState powerState, final boolean dynamicScalingEnabled, String vmType, final Long rootDiskOfferingId, String sshkeypairs, Volume volume, Snapshot snapshot) throws InsufficientCapacityException {
         UserVmVO vm = new UserVmVO(id, instanceName, displayName, template.getId(), hypervisorType, template.getGuestOSId(), offering.isOfferHA(),
                 offering.getLimitCpuUse(), owner.getDomainId(), owner.getId(), userId, offering.getId(), userData, userDataId, userDataDetails, hostName);
         vm.setUuid(uuidName);
@@ -4744,7 +4770,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
             orchestrateVirtualMachineCreate(vm, guestOSCategory, computeTags, rootDiskTags, plan, rootDiskSize, template, hostName, displayName, owner,
                     diskOfferingId, diskSize, offering, isIso,networkNicMap, hypervisorType, extraDhcpOptionMap, dataDiskTemplateToDiskOfferingMap,
-                    rootDiskOfferingId);
+                    rootDiskOfferingId, volume, snapshot);
 
         }
         CallContext.current().setEventDetails("Vm Id: " + vm.getUuid());
@@ -4777,16 +4803,16 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                         ServiceOffering offering, boolean isIso, LinkedHashMap<String, List<NicProfile>> networkNicMap,
                                         HypervisorType hypervisorType,
                                         Map<String, Map<Integer, String>> extraDhcpOptionMap, Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                        Long rootDiskOfferingId) throws InsufficientCapacityException{
+                                        Long rootDiskOfferingId, Volume volume, Snapshot snapshot) throws InsufficientCapacityException{
         try {
             if (isIso) {
                 _orchSrvc.createVirtualMachineFromScratch(vm.getUuid(), Long.toString(owner.getAccountId()), vm.getIsoId().toString(), hostName, displayName,
                         hypervisorType.name(), guestOSCategory.getName(), offering.getCpu(), offering.getSpeed(), offering.getRamSize(), diskSize, computeTags, rootDiskTags,
-                        networkNicMap, plan, extraDhcpOptionMap, rootDiskOfferingId);
+                        networkNicMap, plan, extraDhcpOptionMap, rootDiskOfferingId, volume, snapshot);
             } else {
                 _orchSrvc.createVirtualMachine(vm.getUuid(), Long.toString(owner.getAccountId()), Long.toString(template.getId()), hostName, displayName, hypervisorType.name(),
                         offering.getCpu(), offering.getSpeed(), offering.getRamSize(), diskSize, computeTags, rootDiskTags, networkNicMap, plan, rootDiskSize, extraDhcpOptionMap,
-                        dataDiskTemplateToDiskOfferingMap, diskOfferingId, rootDiskOfferingId);
+                        dataDiskTemplateToDiskOfferingMap, diskOfferingId, rootDiskOfferingId, volume, snapshot);
             }
 
             if (logger.isDebugEnabled()) {
@@ -4908,13 +4934,13 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                                   final long accountId, final long userId, final ServiceOfferingVO offering, final boolean isIso, final String sshPublicKeys, final LinkedHashMap<String, List<NicProfile>> networkNicMap,
                                   final long id, final String instanceName, final String uuidName, final HypervisorType hypervisorType, final Map<String, String> customParameters, final Map<String,
             Map<Integer, String>> extraDhcpOptionMap, final Map<Long, DiskOffering> dataDiskTemplateToDiskOfferingMap,
-                                  Map<String, String> userVmOVFPropertiesMap, final boolean dynamicScalingEnabled, String vmType, final Long rootDiskOfferingId, String sshkeypairs) throws InsufficientCapacityException {
+                                  Map<String, String> userVmOVFPropertiesMap, final boolean dynamicScalingEnabled, String vmType, final Long rootDiskOfferingId, String sshkeypairs, Volume volume, Snapshot snapshot) throws InsufficientCapacityException {
         return commitUserVm(false, zone, null, null, template, hostName, displayName, owner,
                 diskOfferingId, diskSize, userData, userDataId, userDataDetails, isDisplayVm, keyboard,
                 accountId, userId, offering, isIso, sshPublicKeys, networkNicMap,
                 id, instanceName, uuidName, hypervisorType, customParameters,
                 extraDhcpOptionMap, dataDiskTemplateToDiskOfferingMap,
-                userVmOVFPropertiesMap, null, dynamicScalingEnabled, vmType, rootDiskOfferingId, sshkeypairs);
+                userVmOVFPropertiesMap, null, dynamicScalingEnabled, vmType, rootDiskOfferingId, sshkeypairs, volume, snapshot);
     }
 
     public void validateRootDiskResize(final HypervisorType hypervisorType, Long rootDiskSize, VMTemplateVO templateVO, UserVmVO vm, final Map<String, String> customParameters) throws InvalidParameterValueException
@@ -6134,11 +6160,40 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             }
         }
 
+        Account caller = CallContext.current().getCallingAccount();
+        Long callerId = caller.getId();
+
         Long templateId = cmd.getTemplateId();
+        VolumeInfo volume = null;
+        SnapshotVO snapshot = null;
+
+        if (cmd.getVolumeId() != null) {
+            volume = getVolume(cmd.getVolumeId(), templateId, false);
+            if (volume == null) {
+                throw new InvalidParameterValueException("Could not find volume with id=" + cmd.getVolumeId());
+            }
+            _accountMgr.checkAccess(caller, null, true, volume);
+            templateId = volume.getTemplateId();
+            overrideDiskOfferingId = volume.getDiskOfferingId();
+        } else if (cmd.getSnapshotId() != null) {
+            snapshot = _snapshotDao.findById(cmd.getSnapshotId());
+            if (snapshot == null) {
+                throw new InvalidParameterValueException("Could not find snapshot with id=" + cmd.getSnapshotId());
+            }
+            _accountMgr.checkAccess(caller, null, true, snapshot);
+            VolumeInfo volumeOfSnapshot = getVolume(snapshot.getVolumeId(), templateId, true);
+            templateId = volumeOfSnapshot.getTemplateId();
+            overrideDiskOfferingId = volumeOfSnapshot.getDiskOfferingId();
+        }
 
         boolean dynamicScalingEnabled = cmd.isDynamicScalingEnabled();
 
-        VirtualMachineTemplate template = _entityMgr.findById(VirtualMachineTemplate.class, templateId);
+        VirtualMachineTemplate template = null;
+        if (volume != null || snapshot != null) {
+            template = _entityMgr.findByIdIncludingRemoved(VirtualMachineTemplate.class, templateId);
+        } else {
+            template = _entityMgr.findById(VirtualMachineTemplate.class, templateId);
+        }
         // Make sure a valid template ID was specified
         if (template == null) {
             throw new InvalidParameterValueException("Unable to use template " + templateId);
@@ -6149,6 +6204,10 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             throw new InvalidParameterValueException("Can't deploy VNF appliance from a non-VNF template");
         }
 
+        if (cmd.isVolumeOrSnapshotProvided() &&
+                (!(HypervisorType.KVM.equals(template.getHypervisorType()) || HypervisorType.KVM.equals(cmd.getHypervisor())))) {
+            throw new InvalidParameterValueException("Deploying a virtual machine with existing volume/snapshot is supported only from KVM hypervisors");
+        }
         ServiceOfferingJoinVO svcOffering = serviceOfferingJoinDao.findById(serviceOfferingId);
 
         if (template.isDeployAsIs()) {
@@ -6208,9 +6267,6 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         userData = finalizeUserData(userData, userDataId, template);
         userData = userDataManager.validateUserData(userData, cmd.getHttpMethod());
 
-        Account caller = CallContext.current().getCallingAccount();
-        Long callerId = caller.getId();
-
         boolean isRootAdmin = _accountService.isRootAdmin(callerId);
 
         Long hostId = cmd.getHostId();
@@ -6237,7 +6293,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                 vm = createBasicSecurityGroupVirtualMachine(zone, serviceOffering, template, getSecurityGroupIdList(cmd, zone, template, owner), owner, name, displayName, diskOfferingId,
                         size , group , cmd.getHypervisor(), cmd.getHttpMethod(), userData, userDataId, userDataDetails, sshKeyPairNames, cmd.getIpToNetworkMap(), addrs, displayVm , keyboard , cmd.getAffinityGroupIdList(),
                         cmd.getDetails(), cmd.getCustomId(), cmd.getDhcpOptionsMap(),
-                        dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, overrideDiskOfferingId);
+                        dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, overrideDiskOfferingId, volume, snapshot);
             }
         } else {
             if (_networkModel.checkSecurityGroupSupportForNetwork(owner, zone, networkIds,
@@ -6245,7 +6301,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                 vm = createAdvancedSecurityGroupVirtualMachine(zone, serviceOffering, template, networkIds, getSecurityGroupIdList(cmd, zone, template, owner), owner, name,
                         displayName, diskOfferingId, size, group, cmd.getHypervisor(), cmd.getHttpMethod(), userData, userDataId, userDataDetails, sshKeyPairNames, cmd.getIpToNetworkMap(), addrs, displayVm, keyboard,
                         cmd.getAffinityGroupIdList(), cmd.getDetails(), cmd.getCustomId(), cmd.getDhcpOptionsMap(),
-                        dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, overrideDiskOfferingId, null);
+                        dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, overrideDiskOfferingId, null, volume, snapshot);
 
             } else {
                 if (cmd.getSecurityGroupIdList() != null && !cmd.getSecurityGroupIdList().isEmpty()) {
@@ -6253,7 +6309,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                 }
                 vm = createAdvancedVirtualMachine(zone, serviceOffering, template, networkIds, owner, name, displayName, diskOfferingId, size, group,
                         cmd.getHypervisor(), cmd.getHttpMethod(), userData, userDataId, userDataDetails, sshKeyPairNames, cmd.getIpToNetworkMap(), addrs, displayVm, keyboard, cmd.getAffinityGroupIdList(), cmd.getDetails(),
-                        cmd.getCustomId(), cmd.getDhcpOptionsMap(), dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, null, overrideDiskOfferingId);
+                        cmd.getCustomId(), cmd.getDhcpOptionsMap(), dataDiskTemplateToDiskOfferingMap, userVmOVFProperties, dynamicScalingEnabled, null, overrideDiskOfferingId, volume, snapshot);
                 if (cmd instanceof DeployVnfApplianceCmd) {
                     vnfTemplateManager.createIsolatedNetworkRulesForVnfAppliance(zone, template, owner, vm, (DeployVnfApplianceCmd) cmd);
                 }
@@ -6402,6 +6458,30 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         logger.debug("Instance lease for instanceId: {} is configured to expire on: {} with action: {}", vm.getUuid(), formattedLeaseExpiryDate, leaseExpiryAction);
     }
 
+    private VolumeInfo getVolume(long id, Long templateId, boolean isSnapshot) {
+        VolumeInfo volume = volFactory.getVolume(id);
+        if (volume != null) {
+            if (volume.getDataStore() == null || !ScopeType.ZONE.equals(volume.getDataStore().getScope().getScopeType())) {
+                throw new InvalidParameterValueException("Deployment of virtual machine is supported only for Zone-wide storage pools");
+            }
+            checkIfVolumeTemplateIsTheSameAsTheProvided(volume, templateId);
+            if (volume.getInstanceId() != null && !isSnapshot) {
+                throw new InvalidParameterValueException(String.format("The volume %s is already attached to a VM %s", volume, volume.getInstanceId()));
+            }
+        }
+        return volume;
+    }
+
+    private void checkIfVolumeTemplateIsTheSameAsTheProvided(VolumeInfo volume, Long templateId) {
+        if (volume.getTemplateId() != null) {
+            if (templateId != null && !volume.getTemplateId().equals(templateId)) {
+                throw new InvalidParameterValueException(String.format("The volume's template %s is not the same as the provided one %s", volume.getTemplateId(), templateId));
+            }
+        } else {
+            throw new InvalidParameterValueException("The provided volume/snapshot doesn't have a template to deploy a VM");
+        }
+    }
+
     /**
      * Persist extra configuration data in the user_vm_details table as key/value pair
      * @param decodedUrl String consisting of the extra config data to appended onto the vmx file for VMware instances
@@ -9193,7 +9273,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                 null, null, userData, null, null, isDisplayVm, keyboard,
                 accountId, userId, serviceOffering, template.getFormat().equals(ImageFormat.ISO), sshPublicKeys, networkNicMap,
                 id, instanceName, uuidName, hypervisorType, customParameters,
-                null, null, null, powerState, dynamicScalingEnabled, null, serviceOffering.getDiskOfferingId(), null);
+                null, null, null, powerState, dynamicScalingEnabled, null, serviceOffering.getDiskOfferingId(), null, null, null);
     }
 
     @Override

server/src/test/java/com/cloud/network/as/AutoScaleManagerImplTest.java

@@ -1273,7 +1273,7 @@ public class AutoScaleManagerImplTest {
         when(zoneMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Basic);
         when(userVmService.createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), eq(true), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any())).thenReturn(userVmMock);
+                any(), any(), any(), any(), eq(true), any(), any(), any())).thenReturn(userVmMock);
 
         UserVm result = autoScaleManagerImplSpy.createNewVM(asVmGroupMock);
 
@@ -1284,7 +1284,7 @@ public class AutoScaleManagerImplTest {
         Mockito.verify(userVmService).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(),
                 matches(vmHostNamePattern), matches(vmHostNamePattern),
                 any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), eq(true), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any());
+                any(), any(), any(), any(), eq(true), any(), any(), any());
         Mockito.verify(asVmGroupMock).setNextVmSeq(nextVmSeq + 1);
     }
 
@@ -1320,7 +1320,7 @@ public class AutoScaleManagerImplTest {
         when(zoneMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Advanced);
         when(userVmService.createAdvancedSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), any(), any(), any(),
-                any(), any(), any(), any(), any(), eq(true), any(), any())).thenReturn(userVmMock);
+                any(), any(), any(), any(), any(), eq(true), any(), any(), any(), any())).thenReturn(userVmMock);
         when(networkModel.checkSecurityGroupSupportForNetwork(account, zoneMock,
                 List.of(networkId), Collections.emptyList())).thenReturn(true);
 
@@ -1333,7 +1333,7 @@ public class AutoScaleManagerImplTest {
         Mockito.verify(userVmService).createAdvancedSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(),
                 matches(vmHostNamePattern), matches(vmHostNamePattern),
                 any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), any(), any(), any(),
-                any(), any(), any(), any(), any(), eq(true), any(), any());
+                any(), any(), any(), any(), any(), eq(true), any(), any(), any(), any());
         Mockito.verify(asVmGroupMock).setNextVmSeq(nextVmSeq + 2);
     }
 
@@ -1369,7 +1369,7 @@ public class AutoScaleManagerImplTest {
         when(zoneMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Advanced);
         when(userVmService.createAdvancedVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), eq(true), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any(), any())).thenReturn(userVmMock);
+                any(), any(), any(), any(), eq(true), any(), any(), any(), any())).thenReturn(userVmMock);
         when(networkModel.checkSecurityGroupSupportForNetwork(account, zoneMock,
                 List.of(networkId), Collections.emptyList())).thenReturn(false);
 
@@ -1382,7 +1382,7 @@ public class AutoScaleManagerImplTest {
         Mockito.verify(userVmService).createAdvancedVirtualMachine(any(), any(), any(), any(), any(),
                 matches(vmHostNamePattern), matches(vmHostNamePattern),
                 any(), any(), any(), any(), any(), eq(userData), eq(userDataId), eq(userDataDetails.toString()), any(), any(), any(), eq(true), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any(), any());
+                any(), any(), any(), any(), eq(true), any(), any(), any(), any());
         Mockito.verify(asVmGroupMock).setNextVmSeq(nextVmSeq + 3);
     }
 

server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java

@@ -54,6 +54,8 @@ import org.apache.cloudstack.api.command.user.vm.UpdateVMCmd;
 import org.apache.cloudstack.api.command.user.volume.ResizeVolumeCmd;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
+import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
+import org.apache.cloudstack.engine.subsystem.api.storage.Scope;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.template.VnfTemplateManager;
@@ -156,7 +158,9 @@ import com.cloud.vm.dao.UserVmDao;
 import com.cloud.vm.dao.UserVmDetailsDao;
 import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
+
 import org.apache.cloudstack.vm.lease.VMLeaseManager;
+
 import org.mockito.MockedStatic;
 
 import java.text.SimpleDateFormat;
@@ -174,6 +178,9 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.Mockito.verify;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+
 @RunWith(MockitoJUnitRunner.class)
 public class UserVmManagerImplTest {
 
@@ -382,11 +389,28 @@ public class UserVmManagerImplTest {
     @Mock
     StorageManager storageManager;
 
+    @Mock
+    private VolumeDataFactory volumeDataFactory;
+
+    @Mock
+    private VolumeInfo volumeInfo;
+
+    @Mock
+    private SnapshotVO snapshotMock;
+
+    @Mock
+    private PrimaryDataStore primaryDataStore;
+
+    @Mock
+    private Scope scopeMock;
+
     private static final long vmId = 1l;
     private static final long zoneId = 2L;
     private static final long accountId = 3L;
     private static final long serviceOfferingId = 10L;
     private static final long templateId = 11L;
+    private static final long volumeId = 1L;
+    private static final long snashotId = 1L;
 
     private static final long GiB_TO_BYTES = 1024 * 1024 * 1024;
 
@@ -1100,14 +1124,14 @@ public class UserVmManagerImplTest {
         when(_dcMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Basic);
         Mockito.doReturn(userVmVoMock).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any());
+                any(), any(), any(), any(), eq(true), any(), any(), any());
 
         UserVm result = userVmManagerImpl.createVirtualMachine(deployVMCmd);
         assertEquals(userVmVoMock, result);
         Mockito.verify(vnfTemplateManager).validateVnfApplianceNics(templateMock, null);
         Mockito.verify(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any());
+                any(), any(), any(), any(), eq(true), any(), any(), any());
     }
 
     private List<VolumeVO> mockVolumesForIsAnyVmVolumeUsingLocalStorageTest(int localVolumes, int nonLocalVolumes) {
@@ -1360,7 +1384,7 @@ public class UserVmManagerImplTest {
 
         Mockito.doThrow(cre).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
-                any(), any(), any(), any(), eq(true), any());
+                any(), any(), any(), any(), eq(true), any(), any(), any());
 
         CloudRuntimeException creThrown = assertThrows(CloudRuntimeException.class, () -> userVmManagerImpl.createVirtualMachine(deployVMCmd));
         ArrayList<ExceptionProxyObject> proxyIdList = creThrown.getIdProxyList();
@@ -3210,6 +3234,7 @@ public class UserVmManagerImplTest {
 
     @Test
     public void validateNullStorageAccessGroupsOnSrcHost() {
+
         Host srcHost = Mockito.mock(Host.class);
         Host destHost = Mockito.mock(Host.class);
 
@@ -3390,9 +3415,84 @@ public class UserVmManagerImplTest {
     }
 
     Map<String, String> getLeaseDetails(int leaseDuration, String leaseExecution) {
+
         Map<String, String> leaseDetails = new HashMap<>();
         leaseDetails.put(VmDetailConstants.INSTANCE_LEASE_EXPIRY_DATE, getLeaseExpiryDate(leaseDuration));
         leaseDetails.put(VmDetailConstants.INSTANCE_LEASE_EXECUTION, leaseExecution);
         return leaseDetails;
     }
+
+    @Test
+    public void createVirtualMachineWithExistingVolume() throws ResourceUnavailableException, InsufficientCapacityException, ResourceAllocationException {
+        DeployVMCmd deployVMCmd = new DeployVMCmd();
+        ReflectionTestUtils.setField(deployVMCmd, "zoneId", zoneId);
+        ReflectionTestUtils.setField(deployVMCmd, "serviceOfferingId", serviceOfferingId);
+        ReflectionTestUtils.setField(deployVMCmd, "volumeId", volumeId);
+        deployVMCmd._accountService = accountService;
+
+        when(accountService.finalyzeAccountId(nullable(String.class), nullable(Long.class), nullable(Long.class), eq(true))).thenReturn(accountId);
+        when(accountService.getActiveAccountById(accountId)).thenReturn(account);
+        when(entityManager.findById(DataCenter.class, zoneId)).thenReturn(_dcMock);
+        when(entityManager.findById(ServiceOffering.class, serviceOfferingId)).thenReturn(serviceOffering);
+        when(entityManager.findById(DiskOffering.class, serviceOffering.getId())).thenReturn(smallerDisdkOffering);
+        when(entityManager.findByIdIncludingRemoved(VirtualMachineTemplate.class, templateId)).thenReturn(templateMock);
+        when(volumeDataFactory.getVolume(volumeId)).thenReturn(volumeInfo);
+        when(volumeInfo.getTemplateId()).thenReturn(templateId);
+        when(volumeInfo.getInstanceId()).thenReturn(null);
+        when(volumeInfo.getDataStore()).thenReturn(primaryDataStore);
+        when(primaryDataStore.getScope()).thenReturn(scopeMock);
+        when(primaryDataStore.getScope().getScopeType()).thenReturn(ScopeType.ZONE);
+        when(templateMock.getTemplateType()).thenReturn(Storage.TemplateType.VNF);
+        when(templateMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        when(templateMock.isDeployAsIs()).thenReturn(false);
+        when(templateMock.getFormat()).thenReturn(Storage.ImageFormat.QCOW2);
+        when(templateMock.getUserDataId()).thenReturn(null);
+        Mockito.doNothing().when(vnfTemplateManager).validateVnfApplianceNics(any(), nullable(List.class));
+        when(_dcMock.isLocalStorageEnabled()).thenReturn(false);
+        when(_dcMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Basic);
+        Mockito.doReturn(userVmVoMock).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
+                any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
+                any(), any(), any(), any(), eq(true), any(), any(), any());
+
+
+        userVmManagerImpl.createVirtualMachine(deployVMCmd);
+    }
+
+    @Test
+    public void createVirtualMachineWithExistingSnapshot() throws ResourceUnavailableException, InsufficientCapacityException, ResourceAllocationException {
+        DeployVMCmd deployVMCmd = new DeployVMCmd();
+        ReflectionTestUtils.setField(deployVMCmd, "zoneId", zoneId);
+        ReflectionTestUtils.setField(deployVMCmd, "serviceOfferingId", serviceOfferingId);
+        ReflectionTestUtils.setField(deployVMCmd, "snapshotId", snashotId);
+        deployVMCmd._accountService = accountService;
+
+        when(accountService.finalyzeAccountId(nullable(String.class), nullable(Long.class), nullable(Long.class), eq(true))).thenReturn(accountId);
+        when(accountService.getActiveAccountById(accountId)).thenReturn(account);
+        when(entityManager.findById(DataCenter.class, zoneId)).thenReturn(_dcMock);
+        when(entityManager.findById(ServiceOffering.class, serviceOfferingId)).thenReturn(serviceOffering);
+        when(entityManager.findById(DiskOffering.class, serviceOffering.getId())).thenReturn(smallerDisdkOffering);
+        when(snapshotDaoMock.findById(snashotId)).thenReturn(snapshotMock);
+        when(entityManager.findByIdIncludingRemoved(VirtualMachineTemplate.class, templateId)).thenReturn(templateMock);
+        when(volumeDataFactory.getVolume(volumeId)).thenReturn(volumeInfo);
+        when(snapshotMock.getVolumeId()).thenReturn(volumeId);
+        when(volumeInfo.getTemplateId()).thenReturn(templateId);
+        when(volumeInfo.getInstanceId()).thenReturn(null);
+        when(volumeInfo.getDataStore()).thenReturn(primaryDataStore);
+        when(primaryDataStore.getScope()).thenReturn(scopeMock);
+        when(primaryDataStore.getScope().getScopeType()).thenReturn(ScopeType.ZONE);
+        when(templateMock.getTemplateType()).thenReturn(Storage.TemplateType.VNF);
+        when(templateMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        when(templateMock.isDeployAsIs()).thenReturn(false);
+        when(templateMock.getFormat()).thenReturn(Storage.ImageFormat.QCOW2);
+        when(templateMock.getUserDataId()).thenReturn(null);
+        Mockito.doNothing().when(vnfTemplateManager).validateVnfApplianceNics(any(), nullable(List.class));
+        when(_dcMock.isLocalStorageEnabled()).thenReturn(false);
+        when(_dcMock.getNetworkType()).thenReturn(DataCenter.NetworkType.Basic);
+        Mockito.doReturn(userVmVoMock).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
+                any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
+                any(), any(), any(), any(), eq(true), any(), any(), any());
+
+
+        userVmManagerImpl.createVirtualMachine(deployVMCmd);
+    }
 }

services/secondary-storage/controller/src/main/java/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java

@@ -693,7 +693,7 @@ public class SecondaryStorageManagerImpl extends ManagerBase implements Secondar
             secStorageVm = createOrUpdateSecondaryStorageVm(secStorageVm, dataCenterId, id, name, serviceOffering,
                     template, systemAcct, role);
             try {
-                _itMgr.allocate(name, template, serviceOffering, networks, plan, template.getHypervisorType());
+                _itMgr.allocate(name, template, serviceOffering, networks, plan, template.getHypervisorType(), null, null);
                 secStorageVm = _secStorageVmDao.findById(secStorageVm.getId());
                 _itMgr.checkDeploymentPlan(secStorageVm, template, serviceOffering, systemAcct, plan);
                 break;

test/integration/plugins/storpool/test_storpool_tiers.py

@@ -26,7 +26,9 @@ from marvin.lib.base import (DiskOffering,
                              StoragePool,
                              VirtualMachine,
                              SecurityGroup,
-                             ResourceDetails
+                             ResourceDetails,
+                             Snapshot,
+                             Volume,
                              )
 from marvin.lib.common import (get_domain,
                                get_template,
@@ -167,6 +169,29 @@ class TestStorPoolTiers(cloudstackTestCase):
         cls.random_data_0 = random_gen(size=100)
         cls.test_dir = "/tmp"
         cls.random_data = "random.data"
+        cls.virtual_machine = VirtualMachine.create(
+            cls.apiclient,
+            {"name": "StorPool-%s" % uuid.uuid4()},
+            zoneid=cls.zone.id,
+            templateid=cls.template.id,
+            accountid=cls.account.name,
+            domainid=cls.account.domainid,
+            serviceofferingid=cls.service_offering.id,
+            overridediskofferingid=cls.disk_offerings_tier1_tags.id,
+            hypervisor=cls.hypervisor,
+            rootdisksize=10
+        )
+
+        volume = list_volumes(
+            cls.apiclient,
+            virtualmachineid=cls.virtual_machine.id,
+            type='ROOT',
+            listall=True
+        )[0]
+        cls.snapshot = Snapshot.create(
+            cls.apiclient,
+            volume.id,
+        )
         return
 
     @classmethod
@@ -200,6 +225,7 @@ class TestStorPoolTiers(cloudstackTestCase):
     def test_01_check_tags_on_deployed_vm_and_datadisk(self):
         virtual_machine_tier1_tag = self.deploy_vm_and_check_tier_tag()
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_02_change_offering_on_attached_root_disk(self):
@@ -213,6 +239,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     def test_03_change_offering_on_attached_data_disk(self):
         virtual_machine_tier1_tag = self.deploy_vm_and_check_tier_tag()
@@ -225,6 +252,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_04_check_templates_on_deployed_vm_and_datadisk(self):
@@ -246,6 +274,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         for v in volumes:
             self.check_storpool_template(v, self.disk_offerings_tier1_template.id, self.spTemplate)
         virtual_machine_template_tier1.stop(self.apiclient, forced=True)
+        virtual_machine_template_tier1.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_05_check_templates_on_deployed_vm_and_datadisk_tier2(self):
@@ -267,6 +296,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         for v in volumes:
             self.check_storpool_template(v, self.disk_offerings_tier2_template.id, self.spTemplate)
         virtual_machine_template_tier2.stop(self.apiclient, forced=True)
+        virtual_machine_template_tier2.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_06_change_offerings_with_tags_detached_volume(self):
@@ -300,6 +330,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.changeOfferingForVolume(volumes[0].id, self.disk_offerings_tier1_tags.id, volumes[0].size)
         self.vc_policy_tags(volumes=volumes, vm=virtual_machine_tier2_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier1_tags.id, attached=True)
+        virtual_machine_tier2_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_07_change_offerings_with_template_detached_volume(self):
@@ -332,6 +363,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.changeOfferingForVolume(volumes[0].id, self.disk_offerings_tier1_template.id, volumes[0].size)
         self.check_storpool_template(volume=volumes[0], disk_offering_id=self.disk_offerings_tier1_template.id,
                                      qos_or_template=self.spTemplate)
+        virtual_machine_tier2_template.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_08_deploy_vm_with_tags_and_template_in_offerings(self):
@@ -370,6 +402,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.changeOfferingForVolume(volumes[0].id, self.disk_offerings_tier1_tags.id, volumes[0].size)
         self.vc_policy_tags(volumes=volumes, vm=virtual_machine_tier2_template, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier1_tags.id, attached=True)
+        virtual_machine_tier2_template.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_09_resize_root_volume(self):
@@ -386,6 +419,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_10_shrink_root_volume(self):
@@ -403,6 +437,7 @@ class TestStorPoolTiers(cloudstackTestCase):
                                    listall=True)
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_11_resize_data_volume(self):
@@ -419,6 +454,7 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
 
     @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
     def test_12_shrink_data_volume(self):
@@ -436,6 +472,25 @@ class TestStorPoolTiers(cloudstackTestCase):
         self.vc_policy_tags(volumes=root_volume, vm=virtual_machine_tier1_tag, qos_or_template=self.qos,
                             disk_offering_id=self.disk_offerings_tier2_tags.id, attached=True)
         virtual_machine_tier1_tag.stop(self.apiclient, forced=True)
+        virtual_machine_tier1_tag.delete(self.apiclient, expunge=True)
+
+    @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
+    def test_13_deploy_vm_from_volume_check_tags(self):
+        vm = self.deploy_vm_from_snapshot_or_template(snapshotid=self.snapshot.id, is_snapshot=False)
+        root_volume = list_volumes(self.apiclient, virtualmachineid=vm.id, type="ROOT",
+                               listall=True)
+        self.vc_policy_tags(volumes=root_volume, vm=vm, qos_or_template=self.qos,
+                            disk_offering_id=self.disk_offerings_tier1_tags.id, attached=True)
+        vm.delete(self.apiclient, expunge=True)
+
+    @attr(tags=["advanced", "advancedns", "smoke"], required_hardware="true")
+    def test_14_deploy_vm_from_snapshot_check_tags(self):
+        vm = self.deploy_vm_from_snapshot_or_template(snapshotid=self.snapshot.id, is_snapshot=True)
+        root_volume = list_volumes(self.apiclient, virtualmachineid=vm.id, type="ROOT",
+                                   listall=True)
+        self.vc_policy_tags(volumes=root_volume, vm=vm, qos_or_template=self.qos,
+                            disk_offering_id=self.disk_offerings_tier1_tags.id, attached=True)
+        vm.delete(self.apiclient, expunge=True)
 
     def deploy_vm_and_check_tier_tag(self):
         virtual_machine_tier1_tag = VirtualMachine.create(
@@ -542,3 +597,45 @@ class TestStorPoolTiers(cloudstackTestCase):
         change_offering_for_volume_cmd.shrinkok = shrinkok
 
         return self.apiclient.changeOfferingForVolume(change_offering_for_volume_cmd)
+
+    def deploy_vm_from_snapshot_or_template(self, snapshotid, is_snapshot=False):
+        if is_snapshot:
+            virtual_machine = VirtualMachine.create(self.apiclient,
+                                                    {"name": "StorPool-%s" % uuid.uuid4()},
+                                                    zoneid=self.zone.id,
+                                                    accountid=self.account.name,
+                                                    domainid=self.account.domainid,
+                                                    serviceofferingid=self.service_offering.id,
+                                                    snapshotid=snapshotid,
+                                                    )
+            try:
+                ssh_client = virtual_machine.get_ssh_client()
+            except Exception as e:
+                self.fail("SSH failed for virtual machine: %s - %s" %
+                          (virtual_machine.ipaddress, e))
+
+            return virtual_machine
+        volume = Volume.create_from_snapshot(
+            self.apiclient,
+            snapshot_id=snapshotid,
+            services=self.services,
+            account=self.account.name,
+            domainid=self.account.domainid,
+            disk_offering=self.disk_offerings_tier1_tags.id,
+            zoneid=self.zone.id,
+            size=10
+        )
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "StorPool-%s" % uuid.uuid4()},
+                                                zoneid=self.zone.id,
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                serviceofferingid=self.service_offering.id,
+                                                volumeid=volume.id,
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))
+        return virtual_machine

test/integration/smoke/test_vm_lifecycle_with_snapshot_or_volume.py

@@ -0,0 +1,318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+""" BVT tests for Virtual Machine Life Cycle
+"""
+# Import Local Modules
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.utils import *
+
+from marvin.lib.base import (Account,
+                             Role,
+                             ServiceOffering,
+                             VirtualMachine,
+                             Host,
+                             StoragePool,
+                             Volume,
+                             DiskOffering,
+                             Snapshot,
+                             Template)
+from marvin.lib.common import (get_domain,
+                               get_zone,
+                               get_template,
+                               list_hosts,
+                               list_volumes,
+                               list_storage_pools)
+from marvin.codes import FAILED, PASS
+from nose.plugins.attrib import attr
+
+import uuid
+import unittest
+
+class TestDeployVMFromSnapshotOrVolume(cloudstackTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        testClient = super(TestDeployVMFromSnapshotOrVolume, cls).getClsTestClient()
+        cls.apiclient = testClient.getApiClient()
+        cls.services = testClient.getParsedTestDataConfig()
+
+        # Get Zone, Domain and templates
+        cls.domain = get_domain(cls.apiclient)
+        cls.zone = get_zone(cls.apiclient, testClient.getZoneForTests())
+        cls.services['mode'] = cls.zone.networktype
+        cls.hypervisor = testClient.getHypervisorInfo()
+
+        if cls.hypervisor.lower() != "kvm":
+            raise unittest.SkipTest("Only KVM hypervisor is supported for deployment of a VM with volume/snapshot")
+
+        cls.template = get_template(
+            cls.apiclient,
+            cls.zone.id,
+            account="system"
+        )
+        if cls.template == FAILED:
+            assert False, "get_template failed to return template with description [system]"
+
+        cls.services["small"]["zoneid"] = cls.zone.id
+
+        cls._cleanup = []
+
+        cls.account = Account.create(
+            cls.apiclient,
+            cls.services["account"],
+            domainid=cls.domain.id
+        )
+        cls._cleanup.append(cls.account)
+        cls.debug(cls.account.id)
+
+        storage_pools_response = list_storage_pools(cls.apiclient,
+                                                    zoneid=cls.zone.id,
+                                                    scope="ZONE")
+
+        if storage_pools_response:
+            cls.zone_wide_storage = storage_pools_response[0]
+
+            cls.debug(
+                "zone wide storage id is %s" %
+                cls.zone_wide_storage.id)
+            update1 = StoragePool.update(cls.apiclient,
+                                         id=cls.zone_wide_storage.id,
+                                         tags="test-vm"
+                                         )
+            cls.debug(
+                "Storage %s pool tag%s" %
+                (cls.zone_wide_storage.id, update1.tags))
+            cls.service_offering = ServiceOffering.create(
+                cls.apiclient,
+                cls.services["service_offerings"]["small"],
+                tags="test-vm"
+            )
+            cls._cleanup.append(cls.service_offering)
+
+            do = {
+                "name": "do-tags",
+                "displaytext": "Disk offering with tags",
+                "disksize":8,
+                "tags": "test-vm"
+            }
+            cls.disk_offering = DiskOffering.create(
+                cls.apiclient,
+                do,
+            )
+            cls._cleanup.append(cls.disk_offering)
+        else:
+            raise unittest.SkipTest("No zone wide storage found. Skipping tests")
+
+
+        cls.virtual_machine = VirtualMachine.create(
+            cls.apiclient,
+            cls.services["small"],
+            accountid=cls.account.name,
+            domainid=cls.account.domainid,
+            templateid=cls.template.id,
+            serviceofferingid=cls.service_offering.id,
+            mode=cls.services['mode']
+        )
+        volume = list_volumes(
+            cls.apiclient,
+            virtualmachineid=cls.virtual_machine.id,
+            type='ROOT',
+            listall=True
+        )[0]
+        cls.snapshot = Snapshot.create(
+            cls.apiclient,
+            volume.id,
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestDeployVMFromSnapshotOrVolume, cls).tearDownClass()
+
+    def setUp(self):
+        self.apiclient = self.testClient.getApiClient()
+        self.dbclient = self.testClient.getDbConnection()
+        self.cleanup = []
+
+    def tearDown(self):
+        super(TestDeployVMFromSnapshotOrVolume, self).tearDown()
+
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_01_deploy_vm_with_existing_volume(self):
+        '''
+            Deploy a Virtual machine with existing volume
+        '''
+        self.create_volume_from_snapshot_deploy_vm(self.snapshot.id)
+
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_02_deploy_vm_with_existing_snapshot(self):
+        '''
+            Deploy a Virtual machine with existing snapshot
+        '''
+
+        self.deploy_vm_from_snapshot(self.snapshot)
+
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_03_deploy_vm_with_existing_volume_deleted_template(self):
+        '''
+            Deploy a Virtual machine with existing ROOT volume created from a templated which was deleted
+        '''
+        services = {"displaytext": "Template-1", "name": "Template-1-name", "ostypeid": self.template.ostypeid,
+                    "ispublic": "true"}
+
+        template = Template.create_from_snapshot(self.apiclient, self.snapshot, services)
+        self._cleanup.append(template)
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "Test-%s" % uuid.uuid4()},
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                zoneid=self.zone.id,
+                                                serviceofferingid=self.service_offering.id,
+                                                templateid=template.id,
+                                                mode="basic",
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))
+
+        root_volume = list_volumes(
+            self.apiclient,
+            virtualmachineid=virtual_machine.id,
+            type='ROOT',
+            listall=True
+        )[0]
+        snapshot = Snapshot.create(
+            self.apiclient,
+            root_volume.id,
+        )
+        VirtualMachine.delete(virtual_machine, self.apiclient, expunge=True)
+        self.create_volume_from_snapshot_deploy_vm(snapshot.id)
+
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_04_deploy_vm_with_existing_snapshot_deleted_template(self):
+        '''
+            Deploy a Virtual machine with existing snapshot of a ROOT volume created from a templated which was deleted
+        '''
+        services = {"displaytext": "Template-1", "name": "Template-1-name", "ostypeid": self.template.ostypeid,
+                    "ispublic": "true"}
+
+        template = Template.create_from_snapshot(self.apiclient, self.snapshot, services)
+        self._cleanup.append(template)
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "Test-%s" % uuid.uuid4()},
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                zoneid=self.zone.id,
+                                                serviceofferingid=self.service_offering.id,
+                                                templateid=template.id,
+                                                mode="basic",
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))
+
+        root_volume = list_volumes(
+            self.apiclient,
+            virtualmachineid=virtual_machine.id,
+            type='ROOT',
+            listall=True
+        )[0]
+        snapshot = Snapshot.create(
+            self.apiclient,
+            root_volume.id,
+        )
+        VirtualMachine.delete(virtual_machine, self.apiclient, expunge=True)
+        self.deploy_vm_from_snapshot(snapshot)
+
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_05_deploy_vm_with_existing_snapshot_deleted_volume(self):
+        '''
+            Deploy a Virtual machine with existing snapshot of a ROOT volume which was deleted
+        '''
+
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "Test-%s" % uuid.uuid4()},
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                zoneid=self.zone.id,
+                                                serviceofferingid=self.service_offering.id,
+                                                templateid=self.template.id,
+                                                mode="basic",
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))
+
+        root_volume = list_volumes(
+            self.apiclient,
+            virtualmachineid=virtual_machine.id,
+            type='ROOT',
+            listall=True
+        )[0]
+        snapshot = Snapshot.create(
+            self.apiclient,
+            root_volume.id,
+        )
+        VirtualMachine.delete(virtual_machine, self.apiclient, expunge=True)
+        self.deploy_vm_from_snapshot(snapshot)
+
+    def deploy_vm_from_snapshot(self, snapshot):
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "Test-%s" % uuid.uuid4()},
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                zoneid=self.zone.id,
+                                                serviceofferingid=self.service_offering.id,
+                                                snapshotid=snapshot.id,
+                                                mode="basic",
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))
+
+    def create_volume_from_snapshot_deploy_vm(self, snapshotid):
+        volume = Volume.create_from_snapshot(
+            self.apiclient,
+            snapshot_id=snapshotid,
+            services=self.services,
+            disk_offering=self.disk_offering.id,
+            account=self.account.name,
+            domainid=self.account.domainid,
+            zoneid=self.zone.id,
+        )
+        virtual_machine = VirtualMachine.create(self.apiclient,
+                                                {"name": "Test-%s" % uuid.uuid4()},
+                                                accountid=self.account.name,
+                                                domainid=self.account.domainid,
+                                                zoneid=self.zone.id,
+                                                serviceofferingid=self.service_offering.id,
+                                                volumeid=volume.id,
+                                                mode="basic",
+                                                )
+        try:
+            ssh_client = virtual_machine.get_ssh_client()
+        except Exception as e:
+            self.fail("SSH failed for virtual machine: %s - %s" %
+                      (virtual_machine.ipaddress, e))

tools/marvin/marvin/lib/base.py

@@ -528,7 +528,8 @@ class VirtualMachine:
                rootdiskcontroller=None, vpcid=None, macaddress=None, datadisktemplate_diskoffering_list={},
                properties=None, nicnetworklist=None, bootmode=None, boottype=None, dynamicscalingenabled=None,
                userdataid=None, userdatadetails=None, extraconfig=None, size=None, overridediskofferingid=None,
-               leaseduration=None, leaseexpiryaction=None):
+               leaseduration=None, leaseexpiryaction=None, volumeid=None, snapshotid=None):
+
         """Create the instance"""
 
         cmd = deployVirtualMachine.deployVirtualMachineCmd()
@@ -698,6 +699,12 @@ class VirtualMachine:
         if leaseexpiryaction:
             cmd.leaseexpiryaction = leaseexpiryaction
 
+        if volumeid:
+            cmd.volumeid = volumeid
+
+        if snapshotid:
+            cmd.snapshotid = snapshotid
+
         virtual_machine = apiclient.deployVirtualMachine(cmd, method=method)
 
         if 'password' in list(virtual_machine.__dict__.keys()):
@@ -1215,12 +1222,15 @@ class Volume:
 
     @classmethod
     def create_from_snapshot(cls, apiclient, snapshot_id, services,
-                             account=None, domainid=None, projectid=None):
+                             account=None, domainid=None, projectid=None, zoneid=None, disk_offering=None, size=None):
         """Create Volume from snapshot"""
         cmd = createVolume.createVolumeCmd()
         cmd.name = "-".join([services["diskname"], random_gen()])
         cmd.snapshotid = snapshot_id
-        cmd.zoneid = services["zoneid"]
+        if zoneid:
+            cmd.zoneid = zoneid
+        elif "zoneid" in services:
+            cmd.zoneid = services["zoneid"]
         if "size" in services:
             cmd.size = services["size"]
         if "ispublic" in services:
@@ -1239,6 +1249,12 @@ class Volume:
         if projectid:
             cmd.projectid = projectid
 
+        if disk_offering:
+            cmd.diskofferingid = disk_offering
+
+        if size:
+            cmd.size = size
+
         return Volume(apiclient.createVolume(cmd).__dict__)
 
     @classmethod
@@ -2736,10 +2752,13 @@ class DiskOffering:
         self.__dict__.update(items)
 
     @classmethod
-    def create(cls, apiclient, services, tags=None, custom=False, domainid=None, cacheMode=None, **kwargs):
+    def create(cls, apiclient, services, tags=None, custom=False, domainid=None, cacheMode=None, displaytext=None, **kwargs):
         """Create Disk offering"""
         cmd = createDiskOffering.createDiskOfferingCmd()
-        cmd.displaytext = services["displaytext"]
+        if displaytext:
+            cmd.displaytext = displaytext
+        else:
+            cmd.displaytext = services["displaytext"]
         cmd.name = services["name"]
         if custom:
             cmd.customized = True