Migrate volume improvements, to bypass secondary storage when copy volume between pools is allowed directly (#11625)

* Migrate volume improvements, to bypass secondary storage when copy volume between pools is allowed directly

* Bypass secondary storage for copy volume between zone-wide pools and
- local storage on host in the same zone
- cluser-wide pools in the same zone

* Bypass secondary storage for volumes on ceph/rdb pool when the scope permits

* Fix dest disk format while migrating volume from ceph/rbd to nfs, and some code improvements

* unit tests

* Update suitable disk offering(s) for volume(s) after migrate VM with volumes when change in pool type (shared or local)

Currently, Migrate VM with volume(s) bypasses the service and disk offerings of the volumes, as the target pools for migration are specified,
which ignores the offerings. Offering change is required when pool type (shared or local) is changed, mainly
- when volume on shared pool is migrated to local pool
- when volume on local pool is migrated to shared pool

* Update with proper message while migrate volume when target pool and offering type mismatches (both are not shared/local)

* Consider host scope first during endpoint selection while copying between primary storages

* Update disk offering count (for listDiskOfferings api) while removing offerings with tags mismatch with storage tags

api/src/main/java/com/cloud/offering/DiskOffering.java

@@ -69,6 +69,8 @@ public interface DiskOffering extends InfrastructureEntity, Identity, InternalId
 
     boolean isCustomized();
 
+    boolean isShared();
+
     void setDiskSize(long diskSize);
 
     long getDiskSize();
@@ -99,7 +101,6 @@ public interface DiskOffering extends InfrastructureEntity, Identity, InternalId
 
     Long getBytesReadRateMaxLength();
 
-
     void setBytesWriteRate(Long bytesWriteRate);
 
     Long getBytesWriteRate();
@@ -112,7 +113,6 @@ public interface DiskOffering extends InfrastructureEntity, Identity, InternalId
 
     Long getBytesWriteRateMaxLength();
 
-
     void setIopsReadRate(Long iopsReadRate);
 
     Long getIopsReadRate();
@@ -133,7 +133,6 @@ public interface DiskOffering extends InfrastructureEntity, Identity, InternalId
 
     Long getIopsWriteRateMax();
 
-
     void setIopsWriteRateMaxLength(Long iopsWriteRateMaxLength);
 
     Long getIopsWriteRateMaxLength();

api/src/main/java/com/cloud/storage/VolumeApiService.java

@@ -180,6 +180,8 @@ public interface VolumeApiService {
      */
     boolean doesStoragePoolSupportDiskOfferingTags(StoragePool destPool, String diskOfferingTags);
 
+    boolean validateConditionsToReplaceDiskOfferingOfVolume(Volume volume, DiskOffering newDiskOffering, StoragePool destPool);
+
     Volume destroyVolume(long volumeId, Account caller, boolean expunge, boolean forceExpunge);
 
     void destroyVolume(long volumeId);

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

@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Set;
 
 import com.cloud.exception.ResourceAllocationException;
+import com.cloud.storage.Storage;
 import com.cloud.utils.Pair;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
@@ -182,10 +183,10 @@ public interface VolumeOrchestrationService {
      */
     DiskProfile importVolume(Type type, String name, DiskOffering offering, Long sizeInBytes, Long minIops, Long maxIops,
                              Long zoneId, HypervisorType hypervisorType, VirtualMachine vm, VirtualMachineTemplate template,
-                             Account owner, Long deviceId, Long poolId, String path, String chainInfo);
+                             Account owner, Long deviceId, Long poolId, Storage.StoragePoolType poolType, String path, String chainInfo);
 
     DiskProfile updateImportedVolume(Type type, DiskOffering offering, VirtualMachine vm, VirtualMachineTemplate template,
-                                     Long deviceId, Long poolId, String path, String chainInfo, DiskProfile diskProfile);
+                                     Long deviceId, Long poolId, Storage.StoragePoolType poolType, String path, String chainInfo, DiskProfile diskProfile);
 
     /**
      * Unmanage VM volumes

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/ClusterScope.java

@@ -19,6 +19,7 @@
 package org.apache.cloudstack.engine.subsystem.api.storage;
 
 import com.cloud.storage.ScopeType;
+import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
 
 public class ClusterScope extends AbstractScope {
     private ScopeType type = ScopeType.CLUSTER;
@@ -51,4 +52,9 @@ public class ClusterScope extends AbstractScope {
         return this.zoneId;
     }
 
+    @Override
+    public String toString() {
+        return String.format("ClusterScope %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(
+                this, "zoneId", "clusterId", "podId"));
+    }
 }

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/HostScope.java

@@ -19,8 +19,10 @@
 package org.apache.cloudstack.engine.subsystem.api.storage;
 
 import com.cloud.storage.ScopeType;
+import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
 
 public class HostScope extends AbstractScope {
+    private ScopeType type = ScopeType.HOST;
     private Long hostId;
     private Long clusterId;
     private Long zoneId;
@@ -34,7 +36,7 @@ public class HostScope extends AbstractScope {
 
     @Override
     public ScopeType getScopeType() {
-        return ScopeType.HOST;
+        return this.type;
     }
 
     @Override
@@ -49,4 +51,10 @@ public class HostScope extends AbstractScope {
     public Long getZoneId() {
         return zoneId;
     }
+
+    @Override
+    public String toString() {
+        return String.format("HostScope %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(
+                this, "zoneId", "clusterId", "hostId"));
+    }
 }

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/ZoneScope.java

@@ -19,6 +19,7 @@
 package org.apache.cloudstack.engine.subsystem.api.storage;
 
 import com.cloud.storage.ScopeType;
+import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
 
 public class ZoneScope extends AbstractScope {
     private ScopeType type = ScopeType.ZONE;
@@ -39,4 +40,9 @@ public class ZoneScope extends AbstractScope {
         return this.zoneId;
     }
 
+    @Override
+    public String toString() {
+        return String.format("ZoneScope %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(
+                this, "zoneId"));
+    }
 }

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

@@ -2859,6 +2859,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
             }
             volume.setPath(result.getPath());
             volume.setPoolId(pool.getId());
+            volume.setPoolType(pool.getPoolType());
             if (result.getChainInfo() != null) {
                 volume.setChainInfo(result.getChainInfo());
             }

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

@@ -1423,7 +1423,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
         String volumeToString = getVolumeIdentificationInfos(volume);
 
         VolumeInfo vol = volFactory.getVolume(volume.getId());
-        if (vol == null){
+        if (vol == null) {
             throw new CloudRuntimeException(String.format("Volume migration failed because volume [%s] is null.", volumeToString));
         }
         if (destPool == null) {
@@ -2308,6 +2308,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
                 StoragePoolVO pool = _storagePoolDao.findByUuid(updatedDataStoreUUID);
                 if (pool != null) {
                     vol.setPoolId(pool.getId());
+                    vol.setPoolType(pool.getPoolType());
                 }
             }
             _volsDao.update(volumeId, vol);
@@ -2317,7 +2318,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     @Override
     public DiskProfile importVolume(Type type, String name, DiskOffering offering, Long sizeInBytes, Long minIops, Long maxIops,
                                     Long zoneId, HypervisorType hypervisorType, VirtualMachine vm, VirtualMachineTemplate template, Account owner,
-                                    Long deviceId, Long poolId, String path, String chainInfo) {
+                                    Long deviceId, Long poolId, Storage.StoragePoolType poolType, String path, String chainInfo) {
         if (sizeInBytes == null) {
             sizeInBytes = offering.getDiskSize();
         }
@@ -2358,6 +2359,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
 
         vol.setFormat(getSupportedImageFormatForCluster(hypervisorType));
         vol.setPoolId(poolId);
+        vol.setPoolType(poolType);
         vol.setPath(path);
         vol.setChainInfo(chainInfo);
         vol.setState(Volume.State.Ready);
@@ -2367,7 +2369,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
 
     @Override
     public DiskProfile updateImportedVolume(Type type, DiskOffering offering, VirtualMachine vm, VirtualMachineTemplate template,
-                                    Long deviceId, Long poolId, String path, String chainInfo, DiskProfile diskProfile) {
+                                    Long deviceId, Long poolId, Storage.StoragePoolType poolType, String path, String chainInfo, DiskProfile diskProfile) {
 
         VolumeVO vol = _volsDao.findById(diskProfile.getVolumeId());
         if (vm != null) {
@@ -2401,6 +2403,7 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
 
         vol.setFormat(getSupportedImageFormatForCluster(vm.getHypervisorType()));
         vol.setPoolId(poolId);
+        vol.setPoolType(poolType);
         vol.setPath(path);
         vol.setChainInfo(chainInfo);
         vol.setSize(diskProfile.getSize());

engine/orchestration/src/test/java/org/apache/cloudstack/engine/orchestration/VolumeOrchestratorTest.java

@@ -241,7 +241,7 @@ public class VolumeOrchestratorTest {
 
         volumeOrchestrator.importVolume(volumeType, name, diskOffering, sizeInBytes, null, null,
                 zoneId, hypervisorType, null, null, owner,
-                deviceId, poolId, path, chainInfo);
+                deviceId, poolId, Storage.StoragePoolType.NetworkFilesystem, path, chainInfo);
 
         VolumeVO volume = volumeVOMockedConstructionConstruction.constructed().get(0);
         Mockito.verify(volume, Mockito.never()).setInstanceId(Mockito.anyLong());

engine/schema/src/main/java/com/cloud/storage/DiskOfferingVO.java

@@ -577,11 +577,11 @@ public class DiskOfferingVO implements DiskOffering {
     @Override
     public void setEncrypt(boolean encrypt) { this.encrypt = encrypt; }
 
+    @Override
     public boolean isShared() {
         return !useLocalStorage;
     }
 
-
     public boolean getDiskSizeStrictness() {
         return diskSizeStrictness;
     }

engine/schema/src/main/java/com/cloud/storage/dao/DiskOfferingDao.java

@@ -31,6 +31,8 @@ public interface DiskOfferingDao extends GenericDao<DiskOfferingVO, Long> {
     List<DiskOfferingVO> listAllBySizeAndProvisioningType(long size, Storage.ProvisioningType provisioningType);
 
     List<DiskOfferingVO> findCustomDiskOfferings();
+
     List<DiskOfferingVO> listByStorageTag(String tag);
 
+    List<DiskOfferingVO> listAllActiveAndNonComputeDiskOfferings();
 }

engine/schema/src/main/java/com/cloud/storage/dao/DiskOfferingDaoImpl.java

@@ -26,6 +26,7 @@ import java.util.List;
 import javax.inject.Inject;
 import javax.persistence.EntityExistsException;
 
+import com.cloud.offering.DiskOffering;
 import org.apache.cloudstack.resourcedetail.dao.DiskOfferingDetailsDao;
 import org.springframework.stereotype.Component;
 
@@ -45,6 +46,8 @@ public class DiskOfferingDaoImpl extends GenericDaoBase<DiskOfferingVO, Long> im
     protected DiskOfferingDetailsDao detailsDao;
 
     protected final SearchBuilder<DiskOfferingVO> UniqueNameSearch;
+    protected final SearchBuilder<DiskOfferingVO> ActiveAndNonComputeSearch;
+
     private final String SizeDiskOfferingSearch = "SELECT * FROM disk_offering WHERE " +
             "disk_size = ? AND provisioning_type = ? AND removed IS NULL";
 
@@ -56,6 +59,11 @@ public class DiskOfferingDaoImpl extends GenericDaoBase<DiskOfferingVO, Long> im
         UniqueNameSearch.and("name", UniqueNameSearch.entity().getUniqueName(), SearchCriteria.Op.EQ);
         UniqueNameSearch.done();
 
+        ActiveAndNonComputeSearch = createSearchBuilder();
+        ActiveAndNonComputeSearch.and("state", ActiveAndNonComputeSearch.entity().getState(), SearchCriteria.Op.EQ);
+        ActiveAndNonComputeSearch.and("computeOnly", ActiveAndNonComputeSearch.entity().isComputeOnly(), SearchCriteria.Op.EQ);
+        ActiveAndNonComputeSearch.done();
+
         _computeOnlyAttr = _allAttributes.get("computeOnly");
     }
 
@@ -164,4 +172,12 @@ public class DiskOfferingDaoImpl extends GenericDaoBase<DiskOfferingVO, Long> im
         sc.setParameters("tagEndLike",   "%," + tag);
         return listBy(sc);
     }
+
+    @Override
+    public List<DiskOfferingVO> listAllActiveAndNonComputeDiskOfferings() {
+        SearchCriteria<DiskOfferingVO> sc = ActiveAndNonComputeSearch.create();
+        sc.setParameters("state", DiskOffering.State.Active);
+        sc.setParameters("computeOnly", false);
+        return listBy(sc);
+    }
 }

engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java

@@ -822,6 +822,7 @@ public class VolumeDaoImpl extends GenericDaoBase<VolumeVO, Long> implements Vol
         if (volume.getState() != Volume.State.Destroy) {
             volume.setState(Volume.State.Destroy);
             volume.setPoolId(null);
+            volume.setPoolType(null);
             volume.setInstanceId(null);
             update(volume.getId(), volume);
             remove(volume.getId());

engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/AncientDataMotionStrategy.java

@@ -18,9 +18,11 @@
  */
 package org.apache.cloudstack.storage.motion;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import javax.inject.Inject;
 
@@ -67,6 +69,7 @@ import com.cloud.configuration.Config;
 import com.cloud.host.Host;
 import com.cloud.hypervisor.Hypervisor;
 import com.cloud.storage.DataStoreRole;
+import com.cloud.storage.ScopeType;
 import com.cloud.storage.Snapshot.Type;
 import com.cloud.storage.SnapshotVO;
 import com.cloud.storage.StorageManager;
@@ -85,6 +88,11 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
     protected Logger logger = LogManager.getLogger(getClass());
     private static final String NO_REMOTE_ENDPOINT_SSVM = "No remote endpoint to send command, check if host or ssvm is down?";
     private static final String NO_REMOTE_ENDPOINT_WITH_ENCRYPTION = "No remote endpoint to send command, unable to find a valid endpoint. Requires encryption support: %s";
+    private static final List<StoragePoolType> SUPPORTED_POOL_TYPES_TO_BYPASS_SECONDARY_STORE = Arrays.asList(
+            StoragePoolType.NetworkFilesystem,
+            StoragePoolType.Filesystem,
+            StoragePoolType.RBD
+    );
 
     @Inject
     EndPointSelector selector;
@@ -240,7 +248,6 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
         return dataTO;
     }
 
-
     protected Answer copyObject(DataObject srcData, DataObject destData) {
         return copyObject(srcData, destData, null);
     }
@@ -352,14 +359,12 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
 
         Scope destScope = getZoneScope(destData.getDataStore().getScope());
         DataStore cacheStore = cacheMgr.getCacheStorage(destScope);
-        boolean bypassSecondaryStorage = false;
+        boolean bypassSecondaryStorage = canBypassSecondaryStorage(srcData, destData);
-        if (srcData instanceof VolumeInfo && ((VolumeInfo)srcData).isDirectDownload()) {
-            bypassSecondaryStorage = true;
-        }
         boolean encryptionRequired = anyVolumeRequiresEncryption(srcData, destData);
 
         if (cacheStore == null) {
             if (bypassSecondaryStorage) {
+                logger.debug("Secondary storage is bypassed, copy volume between pools directly");
                 CopyCommand cmd = new CopyCommand(srcData.getTO(), destData.getTO(), _copyvolumewait, VirtualMachineManager.ExecuteInSequence.value());
                 EndPoint ep = selector.select(srcData, destData, encryptionRequired);
                 Answer answer = null;
@@ -388,8 +393,8 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                 answer = copyObject(srcData, objOnImageStore);
 
                 if (answer == null || !answer.getResult()) {
-                    if (answer != null) {
+                    if (answer != null && logger.isDebugEnabled()) {
-                        if (logger.isDebugEnabled()) logger.debug("copy to image store failed: " + answer.getDetails());
+                        logger.debug("copy to image store failed: {}", answer.getDetails());
                     }
                     objOnImageStore.processEvent(Event.OperationFailed);
                     imageStore.delete(objOnImageStore);
@@ -411,8 +416,8 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                 }
 
                 if (answer == null || !answer.getResult()) {
-                    if (answer != null) {
+                    if (answer != null && logger.isDebugEnabled()) {
-                        if (logger.isDebugEnabled()) logger.debug("copy to primary store failed: " + answer.getDetails());
+                        logger.debug("copy to primary store failed: {}", answer.getDetails());
                     }
                     objOnImageStore.processEvent(Event.OperationFailed);
                     imageStore.delete(objOnImageStore);
@@ -423,7 +428,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                     objOnImageStore.processEvent(Event.OperationFailed);
                     imageStore.delete(objOnImageStore);
                 }
-                logger.error("Failed to perform operation: "+ e.getLocalizedMessage());
+                logger.error("Failed to perform operation: {}", e.getLocalizedMessage());
                 throw e;
             }
 
@@ -448,7 +453,78 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
             }
             return answer;
         }
+    }
 
+    private boolean canBypassSecondaryStorage(DataObject srcData, DataObject destData) {
+        if (srcData instanceof VolumeInfo) {
+            if (((VolumeInfo)srcData).isDirectDownload()) {
+                return true;
+            }
+
+            if (destData instanceof VolumeInfo) {
+                Scope srcDataStoreScope = srcData.getDataStore().getScope();
+                Scope destDataStoreScope = destData.getDataStore().getScope();
+                logger.info("srcDataStoreScope: {}, srcData pool type: {}; destDataStoreScope: {}, destData pool type: {}",
+                        srcDataStoreScope, ((VolumeInfo)srcData).getStoragePoolType(), destDataStoreScope, ((VolumeInfo)destData).getStoragePoolType());
+
+                if (srcDataStoreScope != null && destDataStoreScope != null &&
+                        SUPPORTED_POOL_TYPES_TO_BYPASS_SECONDARY_STORE.contains(((VolumeInfo)srcData).getStoragePoolType()) &&
+                        SUPPORTED_POOL_TYPES_TO_BYPASS_SECONDARY_STORE.contains(((VolumeInfo)destData).getStoragePoolType())) {
+
+                    return canDirectlyCopyBetweenDataStoreScopes(srcDataStoreScope, destDataStoreScope);
+                }
+            }
+        }
+
+        return false;
+    }
+
+    private boolean canDirectlyCopyBetweenDataStoreScopes(Scope srcDataStoreScope, Scope destDataStoreScope) {
+        if (srcDataStoreScope == null || destDataStoreScope == null) {
+            return false;
+        }
+
+        if (srcDataStoreScope.isSameScope(destDataStoreScope)) {
+            return true;
+        }
+
+        if (srcDataStoreScope.getScopeType() == ScopeType.HOST) {
+            if (destDataStoreScope.getScopeType() == ScopeType.CLUSTER &&
+                    (Objects.equals(((HostScope) srcDataStoreScope).getClusterId(), ((ClusterScope) destDataStoreScope).getScopeId()))) {
+                return true;
+            }
+            if (destDataStoreScope.getScopeType() == ScopeType.ZONE &&
+                    (Objects.equals(((HostScope) srcDataStoreScope).getZoneId(), ((ZoneScope) destDataStoreScope).getScopeId()))) {
+                return true;
+            }
+        }
+
+        if (destDataStoreScope.getScopeType() == ScopeType.HOST) {
+            if (srcDataStoreScope.getScopeType() == ScopeType.CLUSTER &&
+                    (Objects.equals(((ClusterScope) srcDataStoreScope).getScopeId(), ((HostScope) destDataStoreScope).getClusterId()))) {
+                return true;
+            }
+            if (srcDataStoreScope.getScopeType() == ScopeType.ZONE &&
+                    (Objects.equals(((ZoneScope) srcDataStoreScope).getScopeId(), ((HostScope) destDataStoreScope).getZoneId()))) {
+                return true;
+            }
+        }
+
+        if (srcDataStoreScope.getScopeType() == ScopeType.CLUSTER) {
+            if (destDataStoreScope.getScopeType() == ScopeType.ZONE &&
+                    (Objects.equals(((ClusterScope) srcDataStoreScope).getZoneId(), ((ZoneScope) destDataStoreScope).getScopeId()))) {
+                return true;
+            }
+        }
+
+        if (destDataStoreScope.getScopeType() == ScopeType.CLUSTER) {
+            if (srcDataStoreScope.getScopeType() == ScopeType.ZONE &&
+                    (Objects.equals(((ZoneScope) srcDataStoreScope).getScopeId(), ((ClusterScope) destDataStoreScope).getZoneId()))) {
+                return true;
+            }
+        }
+
+        return false;
     }
 
     protected Answer migrateVolumeToPool(DataObject srcData, DataObject destData) {
@@ -492,6 +568,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
             }
             volumeVo.setPodId(destPool.getPodId());
             volumeVo.setPoolId(destPool.getId());
+            volumeVo.setPoolType(destPool.getPoolType());
             volumeVo.setLastPoolId(oldPoolId);
             // For SMB, pool credentials are also stored in the uri query string.  We trim the query string
             // part  here to make sure the credentials do not get stored in the db unencrypted.

engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/StorageSystemDataMotionStrategy.java

@@ -122,6 +122,7 @@ import com.cloud.storage.VMTemplateStoragePoolVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.Volume;
+import com.cloud.storage.VolumeApiService;
 import com.cloud.storage.VolumeDetailVO;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.DiskOfferingDao;
@@ -194,6 +195,8 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
     @Inject
     private VolumeService _volumeService;
     @Inject
+    public VolumeApiService _volumeApiService;
+    @Inject
     private StorageCacheManager cacheMgr;
     @Inject
     private EndPointSelector selector;
@@ -796,6 +799,7 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
 
         volumeVO.setPodId(destPool.getPodId());
         volumeVO.setPoolId(destPool.getId());
+        volumeVO.setPoolType(destPool.getPoolType());
         volumeVO.setLastPoolId(srcVolumeInfo.getPoolId());
 
         _volumeDao.update(srcVolumeInfo.getId(), volumeVO);
@@ -2348,11 +2352,22 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
                 volumeVO.setFormat(ImageFormat.QCOW2);
                 volumeVO.setLastId(srcVolumeInfo.getId());
 
+                if (Objects.equals(srcVolumeInfo.getDiskOfferingId(), destVolumeInfo.getDiskOfferingId())) {
+                    StoragePoolVO srcPoolVO = _storagePoolDao.findById(srcVolumeInfo.getPoolId());
+                    StoragePoolVO destPoolVO = _storagePoolDao.findById(destVolumeInfo.getPoolId());
+                    if (srcPoolVO != null && destPoolVO != null &&
+                            ((srcPoolVO.isShared() && destPoolVO.isLocal()) || (srcPoolVO.isLocal() && destPoolVO.isShared()))) {
+                        Long offeringId = getSuitableDiskOfferingForVolumeOnPool(volumeVO, destPoolVO);
+                        if (offeringId != null) {
+                            volumeVO.setDiskOfferingId(offeringId);
+                        }
+                    }
+                }
+
                 _volumeDao.update(volumeVO.getId(), volumeVO);
 
                 _volumeService.copyPoliciesBetweenVolumesAndDestroySourceVolumeAfterMigration(Event.OperationSuccessed, null, srcVolumeInfo, destVolumeInfo, false);
 
-
                 // Update the volume ID for snapshots on secondary storage
                 if (!_snapshotDao.listByVolumeId(srcVolumeInfo.getId()).isEmpty()) {
                     _snapshotDao.updateVolumeIds(srcVolumeInfo.getId(), destVolumeInfo.getId());
@@ -2394,17 +2409,32 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
         }
     }
 
+    private Long getSuitableDiskOfferingForVolumeOnPool(VolumeVO volume, StoragePoolVO pool) {
+        List<DiskOfferingVO> diskOfferings = _diskOfferingDao.listAllActiveAndNonComputeDiskOfferings();
+        for (DiskOfferingVO diskOffering : diskOfferings) {
+            try {
+                if (_volumeApiService.validateConditionsToReplaceDiskOfferingOfVolume(volume, diskOffering, pool)) {
+                    logger.debug("Found suitable disk offering {} for the volume {}", diskOffering, volume);
+                    return diskOffering.getId();
+                }
+            } catch (Exception ignore) {
+            }
+        }
+        logger.warn("Unable to find suitable disk offering for the volume {}", volume);
+        return null;
+    }
+
     private VolumeVO duplicateVolumeOnAnotherStorage(Volume volume, StoragePoolVO storagePoolVO) {
         Long lastPoolId = volume.getPoolId();
 
         VolumeVO newVol = new VolumeVO(volume);
-
         newVol.setInstanceId(null);
         newVol.setChainInfo(null);
         newVol.setPath(null);
         newVol.setFolder(null);
         newVol.setPodId(storagePoolVO.getPodId());
         newVol.setPoolId(storagePoolVO.getId());
+        newVol.setPoolType(storagePoolVO.getPoolType());
         newVol.setLastPoolId(lastPoolId);
         newVol.setLastId(volume.getId());
 

engine/storage/datamotion/src/test/java/org/apache/cloudstack/storage/motion/AncientDataMotionStrategyTest.java

@@ -21,20 +21,34 @@ package org.apache.cloudstack.storage.motion;
 import static org.mockito.Mockito.when;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.any;
 
+import com.cloud.storage.Storage;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.StoragePool;
+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.ClusterScope;
+import org.apache.cloudstack.engine.subsystem.api.storage.HostScope;
+import org.apache.cloudstack.engine.subsystem.api.storage.StorageCacheManager;
+import org.apache.cloudstack.engine.subsystem.api.storage.ZoneScope;
 import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.storage.image.store.TemplateObject;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
+import org.apache.cloudstack.storage.volume.VolumeObject;
+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.Mockito;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
 
@@ -57,6 +71,10 @@ public class AncientDataMotionStrategyTest {
     StorageManager storageManager;
     @Mock
     StoragePool storagePool;
+    @Mock
+    StorageCacheManager cacheMgr;
+    @Mock
+    ConfigurationDao configDao;
 
     private static final long POOL_ID = 1l;
     private static final Boolean FULL_CLONE_FLAG = true;
@@ -88,4 +106,186 @@ public class AncientDataMotionStrategyTest {
         verify(dataStoreTO, never()).setFullCloneFlag(any(Boolean.class));
     }
 
+    @Test
+    public void testCanBypassSecondaryStorageForDirectDownload() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        Mockito.doReturn(true).when(srcVolumeInfo).isDirectDownload();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageForUnsupportedDataObject() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+
+        TemplateObject destTemplateInfo = Mockito.spy(new TemplateObject());
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destTemplateInfo);
+        Assert.assertFalse(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageForUnsupportedSrcPoolType() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.PowerFlex).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertFalse(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageForUnsupportedDestPoolType() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.Iscsi).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertFalse(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageWithZoneWideNFSPoolsInSameZone() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageWithClusterWideNFSPoolsInSameCluster() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ClusterScope(5L, 2L, 1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ClusterScope(5L, 2L, 1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageWithLocalAndClusterWideNFSPoolsInSameCluster() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new HostScope(1L, 1L, 1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.Filesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ClusterScope(1L, 1L, 1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+
+        canBypassSecondaryStorage = (boolean) method.invoke(strategy, destVolumeInfo, srcVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageWithLocalAndZoneWideNFSPoolsInSameZone() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new HostScope(1L, 1L, 1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.Filesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+
+        canBypassSecondaryStorage = (boolean) method.invoke(strategy, destVolumeInfo, srcVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
+
+    @Test
+    public void testCanBypassSecondaryStorageWithClusterWideNFSAndZoneWideNFSPoolsInSameZone() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+        VolumeObject srcVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore srcDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ClusterScope(5L, 2L, 1L)).when(srcDataStore).getScope();
+        Mockito.doReturn(srcDataStore).when(srcVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(srcVolumeInfo).getStoragePoolType();
+
+        VolumeObject destVolumeInfo = Mockito.spy(new VolumeObject());
+        DataStore destDataStore = Mockito.mock(DataStore.class);
+        Mockito.doReturn(new ZoneScope(1L)).when(destDataStore).getScope();
+        Mockito.doReturn(destDataStore).when(destVolumeInfo).getDataStore();
+        Mockito.doReturn(Storage.StoragePoolType.NetworkFilesystem).when(destVolumeInfo).getStoragePoolType();
+
+        Method method;
+        method = AncientDataMotionStrategy.class.getDeclaredMethod("canBypassSecondaryStorage", DataObject.class, DataObject.class);
+        method.setAccessible(true);
+        boolean canBypassSecondaryStorage = (boolean) method.invoke(strategy, srcVolumeInfo, destVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+
+        canBypassSecondaryStorage = (boolean) method.invoke(strategy, destVolumeInfo, srcVolumeInfo);
+        Assert.assertTrue(canBypassSecondaryStorage);
+    }
 }

engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/vmsnapshot/DefaultVMSnapshotStrategy.java

@@ -345,6 +345,7 @@ public class DefaultVMSnapshotStrategy extends ManagerBase implements VMSnapshot
                 StoragePool pool = primaryDataStoreDao.findPoolByUUID(volume.getDataStoreUuid());
                 if (pool != null && pool.getId() != volumeVO.getPoolId()) {
                     volumeVO.setPoolId(pool.getId());
+                    volumeVO.setPoolType(pool.getPoolType());
                 }
             }
             if (StringUtils.isNotEmpty(volume.getPath())) {

engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java

@@ -232,7 +232,13 @@ public class DefaultEndPointSelector implements EndPointSelector {
 
         // assumption, at least one of scope should be zone, find the least
         // scope
-        if (srcScope.getScopeType() != ScopeType.ZONE) {
+        if (srcScope.getScopeType() == ScopeType.HOST) {
+            selectedScope = srcScope;
+            poolId = srcStore.getId();
+        } else if (destScope.getScopeType() == ScopeType.HOST) {
+            selectedScope = destScope;
+            poolId = destStore.getId();
+        } else if (srcScope.getScopeType() != ScopeType.ZONE) {
             selectedScope = srcScope;
             poolId = srcStore.getId();
         } else if (destScope.getScopeType() != ScopeType.ZONE) {

engine/storage/volume/src/main/java/org/apache/cloudstack/storage/datastore/PrimaryDataStoreImpl.java

@@ -334,6 +334,7 @@ public class PrimaryDataStoreImpl implements PrimaryDataStore {
             VolumeVO vol = volumeDao.findById(obj.getId());
             if (vol != null) {
                 vol.setPoolId(getId());
+                vol.setPoolType(getPoolType());
                 volumeDao.update(vol.getId(), vol);
             }
         }

engine/storage/volume/src/main/java/org/apache/cloudstack/storage/volume/VolumeDataFactoryImpl.java

@@ -30,6 +30,8 @@ import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
 
@@ -46,6 +48,8 @@ public class VolumeDataFactoryImpl implements VolumeDataFactory {
     DataStoreManager storeMgr;
     @Inject
     VMTemplateDao templateDao;
+    @Inject
+    PrimaryDataStoreDao storagePoolDao;
 
     @Override
     public VolumeInfo getVolume(long volumeId, DataStore store) {
@@ -92,6 +96,10 @@ public class VolumeDataFactoryImpl implements VolumeDataFactory {
             vol = VolumeObject.getVolumeObject(store, volumeVO);
         } else {
             DataStore store = storeMgr.getDataStore(volumeVO.getPoolId(), DataStoreRole.Primary);
+            StoragePoolVO pool = storagePoolDao.findById(volumeVO.getPoolId());
+            if (pool != null) {
+                volumeVO.setPoolType(pool.getPoolType());
+            }
             vol = VolumeObject.getVolumeObject(store, volumeVO);
         }
         if (vol.getTemplateId() != null) {

plugins/hypervisors/hyperv/src/main/java/org/apache/cloudstack/storage/motion/HypervStorageMotionStrategy.java

@@ -155,6 +155,7 @@ public class HypervStorageMotionStrategy implements DataMotionStrategy {
                     volumeVO.setPath(volumeTo.getPath());
                     volumeVO.setPodId(pool.getPodId());
                     volumeVO.setPoolId(pool.getId());
+                    volumeVO.setPoolType(pool.getPoolType());
                     volumeVO.setLastPoolId(oldPoolId);
                     // For SMB, pool credentials are also stored in the uri query string.  We trim the query string
                     // part  here to make sure the credentials do not get stored in the db unencrypted.

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java

@@ -364,16 +364,7 @@ public class KVMStorageProcessor implements StorageProcessor {
                 final TemplateObjectTO newTemplate = new TemplateObjectTO();
                 newTemplate.setPath(primaryVol.getName());
                 newTemplate.setSize(primaryVol.getSize());
-
+                newTemplate.setFormat(getFormat(primaryPool.getType()));
-                if(List.of(
-                    StoragePoolType.RBD,
-                    StoragePoolType.PowerFlex,
-                    StoragePoolType.Linstor,
-                    StoragePoolType.FiberChannel).contains(primaryPool.getType())) {
-                    newTemplate.setFormat(ImageFormat.RAW);
-                } else {
-                    newTemplate.setFormat(ImageFormat.QCOW2);
-                }
                 data = newTemplate;
             } else if (destData.getObjectType() == DataObjectType.VOLUME) {
                 final VolumeObjectTO volumeObjectTO = new VolumeObjectTO();
@@ -2990,7 +2981,7 @@ public class KVMStorageProcessor implements StorageProcessor {
         final VolumeObjectTO srcVol = (VolumeObjectTO)srcData;
         final VolumeObjectTO destVol = (VolumeObjectTO)destData;
         final ImageFormat srcFormat = srcVol.getFormat();
-        final ImageFormat destFormat = destVol.getFormat();
+        ImageFormat destFormat = destVol.getFormat();
         final DataStoreTO srcStore = srcData.getDataStore();
         final DataStoreTO destStore = destData.getDataStore();
         final PrimaryDataStoreTO srcPrimaryStore = (PrimaryDataStoreTO)srcStore;
@@ -3025,33 +3016,35 @@ public class KVMStorageProcessor implements StorageProcessor {
             volume.setFormat(PhysicalDiskFormat.valueOf(srcFormat.toString()));
             volume.setDispName(srcVol.getName());
             volume.setVmName(srcVol.getVmName());
-
+            KVMPhysicalDisk newVolume;
-            String destVolumeName = null;
+            String destVolumeName;
+            destPool = storagePoolMgr.getStoragePool(destPrimaryStore.getPoolType(), destPrimaryStore.getUuid());
             if (destPrimaryStore.isManaged()) {
                 if (!storagePoolMgr.connectPhysicalDisk(destPrimaryStore.getPoolType(), destPrimaryStore.getUuid(), destVolumePath, destPrimaryStore.getDetails())) {
                     logger.warn("Failed to connect dest volume {}, in storage pool {}", destVol, destPrimaryStore);
                 }
                 destVolumeName = derivePath(destPrimaryStore, destData, destPrimaryStore.getDetails());
             } else {
+                PhysicalDiskFormat destPoolDefaultFormat = destPool.getDefaultFormat();
+                destFormat = getFormat(destPoolDefaultFormat);
                 final String volumeName = UUID.randomUUID().toString();
                 destVolumeName = volumeName + "." + destFormat.getFileExtension();
 
                 // Update path in the command for reconciliation
-                if (destData.getPath() == null) {
+                if (StringUtils.isBlank(destVolumePath)) {
                     ((VolumeObjectTO) destData).setPath(destVolumeName);
                 }
             }
 
-            destPool = storagePoolMgr.getStoragePool(destPrimaryStore.getPoolType(), destPrimaryStore.getUuid());
             try {
                 Volume.Type volumeType = srcVol.getVolumeType();
 
                 resource.createOrUpdateLogFileForCommand(cmd, Command.State.PROCESSING_IN_BACKEND);
                 if (srcVol.getPassphrase() != null && (Volume.Type.ROOT.equals(volumeType) || Volume.Type.DATADISK.equals(volumeType))) {
                     volume.setQemuEncryptFormat(QemuObject.EncryptFormat.LUKS);
-                    storagePoolMgr.copyPhysicalDisk(volume, destVolumeName, destPool, cmd.getWaitInMillSeconds(), srcVol.getPassphrase(), destVol.getPassphrase(), srcVol.getProvisioningType());
+                    newVolume = storagePoolMgr.copyPhysicalDisk(volume, destVolumeName, destPool, cmd.getWaitInMillSeconds(), srcVol.getPassphrase(), destVol.getPassphrase(), srcVol.getProvisioningType());
                 } else {
-                    storagePoolMgr.copyPhysicalDisk(volume, destVolumeName, destPool, cmd.getWaitInMillSeconds());
+                    newVolume = storagePoolMgr.copyPhysicalDisk(volume, destVolumeName, destPool, cmd.getWaitInMillSeconds());
                 }
                 resource.createOrUpdateLogFileForCommand(cmd, Command.State.COMPLETED);
             } catch (Exception e) { // Any exceptions while copying the disk, should send failed answer with the error message
@@ -3071,9 +3064,13 @@ public class KVMStorageProcessor implements StorageProcessor {
             }
 
             final VolumeObjectTO newVol = new VolumeObjectTO();
-            String path = destPrimaryStore.isManaged() ? destVolumeName : destVolumePath + File.separator + destVolumeName;
+            String path = destVolumeName;
+            if (!destPrimaryStore.isManaged() && StringUtils.isNotBlank(destVolumePath)) {
+                path = destVolumePath + File.separator + destVolumeName;
+            }
             newVol.setPath(path);
-            newVol.setFormat(destFormat);
+            ImageFormat newVolumeFormat = getFormat(newVolume.getFormat());
+            newVol.setFormat(newVolumeFormat);
             newVol.setEncryptFormat(destVol.getEncryptFormat());
             return new CopyCmdAnswer(newVol);
         } catch (final CloudRuntimeException e) {
@@ -3085,6 +3082,26 @@ public class KVMStorageProcessor implements StorageProcessor {
         }
     }
 
+    private Storage.ImageFormat getFormat(PhysicalDiskFormat format) {
+        if (format == null) {
+            return null;
+        }
+
+        return ImageFormat.valueOf(format.toString().toUpperCase());
+    }
+
+    private Storage.ImageFormat getFormat(StoragePoolType poolType) {
+        if(List.of(
+                StoragePoolType.RBD,
+                StoragePoolType.PowerFlex,
+                StoragePoolType.Linstor,
+                StoragePoolType.FiberChannel).contains(poolType)) {
+            return ImageFormat.RAW;
+        } else {
+            return ImageFormat.QCOW2;
+        }
+    }
+
     /**
      * True if location exists
      */

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/LibvirtStorageAdaptor.java

@@ -1116,7 +1116,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
                 // make room for encryption header on raw format, use LUKS
                 if (format == PhysicalDiskFormat.RAW) {
-                    destFile.setSize(destFile.getSize() - (16<<20));
+                    destFile.setSize(destFile.getSize() - (16 << 20));
                     destFile.setFormat(PhysicalDiskFormat.LUKS);
                 }
 
@@ -1593,7 +1593,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
         String sourcePath = disk.getPath();
 
         KVMPhysicalDisk newDisk;
-        logger.debug("copyPhysicalDisk: disk size:" + toHumanReadableSize(disk.getSize()) + ", virtualsize:" + toHumanReadableSize(disk.getVirtualSize())+" format:"+disk.getFormat());
+        logger.debug("copyPhysicalDisk: disk size:{}, virtualsize:{} format:{}", toHumanReadableSize(disk.getSize()), toHumanReadableSize(disk.getVirtualSize()), disk.getFormat());
         if (destPool.getType() != StoragePoolType.RBD) {
             if (disk.getFormat() == PhysicalDiskFormat.TAR) {
                 newDisk = destPool.createPhysicalDisk(name, PhysicalDiskFormat.DIR, Storage.ProvisioningType.THIN, disk.getVirtualSize(), null);

plugins/hypervisors/vmware/src/main/java/org/apache/cloudstack/storage/motion/VmwareStorageMotionStrategy.java

@@ -433,6 +433,7 @@ public class VmwareStorageMotionStrategy implements DataMotionStrategy {
                     volumeVO.setFolder(pool.getPath());
                     volumeVO.setPodId(pool.getPodId());
                     volumeVO.setPoolId(pool.getId());
+                    volumeVO.setPoolType(pool.getPoolType());
                     volDao.update(volume.getId(), volumeVO);
                     updated = true;
                     break;

plugins/storage/volume/adaptive/src/main/java/org/apache/cloudstack/storage/datastore/driver/AdaptiveDataStoreDriverImpl.java

@@ -857,6 +857,7 @@ public class AdaptiveDataStoreDriverImpl extends CloudStackPrimaryDataStoreDrive
         volumeVO.setPath(finalPath);
         volumeVO.setFormat(ImageFormat.RAW);
         volumeVO.setPoolId(storagePool.getId());
+        volumeVO.setPoolType(storagePool.getPoolType());
         volumeVO.setExternalUuid(managedVolume.getExternalUuid());
         volumeVO.setDisplay(true);
         volumeVO.setDisplayVolume(true);

plugins/storage/volume/default/src/main/java/org/apache/cloudstack/storage/datastore/driver/CloudStackPrimaryDataStoreDriverImpl.java

@@ -505,6 +505,7 @@ public class CloudStackPrimaryDataStoreDriverImpl implements PrimaryDataStoreDri
             StoragePoolVO storagePoolVO = primaryStoreDao.findByUuid(datastoreUUID);
             if (storagePoolVO != null) {
                 volumeVO.setPoolId(storagePoolVO.getId());
+                volumeVO.setPoolType(storagePoolVO.getPoolType());
             } else {
                 logger.warn("Unable to find datastore {} while updating the new datastore of the volume {}", datastoreUUID, vol);
             }

plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java

@@ -868,6 +868,7 @@ public class LinstorPrimaryDataStoreDriverImpl implements PrimaryDataStoreDriver
                     devPath = createVolume(volumeInfo, storagePool);
                     volume.setFolder("/dev/");
                     volume.setPoolId(storagePool.getId());
+                    volume.setPoolType(storagePool.getPoolType());
                     volume.setUuid(vol.getUuid());
                     volume.setPath(vol.getUuid());
 

plugins/storage/volume/storpool/src/main/java/org/apache/cloudstack/storage/motion/StorPoolDataMotionStrategy.java

@@ -425,6 +425,7 @@ public class StorPoolDataMotionStrategy implements DataMotionStrategy {
         newVol.setFolder(null);
         newVol.setPodId(storagePoolVO.getPodId());
         newVol.setPoolId(storagePoolVO.getId());
+        newVol.setPoolType(storagePoolVO.getPoolType());
         newVol.setLastPoolId(lastPoolId);
 
         return _volumeDao.persist(newVol);

server/src/main/java/com/cloud/api/query/QueryManagerImpl.java

@@ -3675,6 +3675,7 @@ public class QueryManagerImpl extends MutualExclusiveIdsManagerBase implements Q
                 String[] offeringTagsArray = (offeringTags == null || offeringTags.isEmpty()) ? new String[0] : offeringTags.split(",");
                 if (!CollectionUtils.isSubCollection(Arrays.asList(requiredTagsArray), Arrays.asList(offeringTagsArray))) {
                     iteratorForTagsChecking.remove();
+                    count--;
                 }
             }
         }

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

@@ -3049,6 +3049,7 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
                     StoragePoolVO storagePoolVO = _storagePoolDao.findByUuid(datastoreName);
                     if (storagePoolVO != null) {
                         volumeVO.setPoolId(storagePoolVO.getId());
+                        volumeVO.setPoolType(storagePoolVO.getPoolType());
                     } else {
                         logger.warn("Unable to find datastore {} while updating the new datastore of the volume {}", datastoreName, volumeVO);
                     }

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

@@ -2937,8 +2937,10 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                 List<StoragePoolVO> childDatastores = _storagePoolDao.listChildStoragePoolsInDatastoreCluster(storageId);
                 Collections.shuffle(childDatastores);
                 volume.setPoolId(childDatastores.get(0).getId());
+                volume.setPoolType(childDatastores.get(0).getPoolType());
             } else {
                 volume.setPoolId(pool.getId());
+                volume.setPoolType(pool.getPoolType());
             }
         }
 
@@ -3225,6 +3227,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     if (storagePoolVO != null) {
                         VolumeVO volumeVO = _volsDao.findById(volumeId);
                         volumeVO.setPoolId(storagePoolVO.getId());
+                        volumeVO.setPoolType(storagePoolVO.getPoolType());
                         _volsDao.update(volumeVO.getId(), volumeVO);
                     } else {
                         logger.warn("Unable to find datastore {} while updating the new datastore of the volume {}", datastoreName, volume);
@@ -3645,12 +3648,16 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
      *
      * If all of the above validations pass, we check if the size of the new disk offering is different from the volume. If it is, we log a warning message.
      */
-    protected void validateConditionsToReplaceDiskOfferingOfVolume(VolumeVO volume, DiskOfferingVO newDiskOffering, StoragePool destPool) {
+    @Override
+    public boolean validateConditionsToReplaceDiskOfferingOfVolume(Volume volume, DiskOffering newDiskOffering, StoragePool destPool) {
         if (newDiskOffering == null) {
-            return;
+            return false;
         }
-        if ((destPool.isShared() && newDiskOffering.isUseLocalStorage()) || destPool.isLocal() && newDiskOffering.isShared()) {
+        if (destPool.isShared() && newDiskOffering.isUseLocalStorage()) {
-            throw new InvalidParameterValueException("You cannot move the volume to a shared storage and assign a disk offering for local storage and vice versa.");
+            throw new InvalidParameterValueException("You cannot move the volume to shared storage, with the disk offering configured for local storage.");
+        }
+        if (destPool.isLocal() && newDiskOffering.isShared()) {
+            throw new InvalidParameterValueException("You cannot move the volume to local storage, with the disk offering configured for shared storage.");
         }
         if (!doesStoragePoolSupportDiskOffering(destPool, newDiskOffering)) {
             throw new InvalidParameterValueException(String.format("Migration failed: target pool [%s, tags:%s] has no matching tags for volume [%s, uuid:%s, tags:%s]", destPool.getName(),
@@ -3675,6 +3682,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     volume, oldDiskOffering, newDiskOffering);
         }
         logger.info("Changing disk offering to [{}] while migrating volume [{}].", newDiskOffering, volume);
+        return true;
     }
 
     /**

server/src/main/java/org/apache/cloudstack/command/ReconcileCommandServiceImpl.java

@@ -1064,6 +1064,7 @@ public class ReconcileCommandServiceImpl extends ManagerBase implements Reconcil
         logger.debug(String.format("Updating volume %s to %s state", sourceVolume, Volume.State.Ready));
         sourceVolume.setState(Volume.State.Ready);
         sourceVolume.setPoolId(srcDataStore.getId());   // restore pool_id and update path
+        sourceVolume.setPoolType(srcDataStore.getPoolType());
         sourceVolume.setPath(srcData.getPath());
         sourceVolume.set_iScsiName(srcData.getPath());
         sourceVolume.setUpdated(new Date());
@@ -1075,6 +1076,7 @@ public class ReconcileCommandServiceImpl extends ManagerBase implements Reconcil
             VolumeVO newVolume = (VolumeVO) newVol;
             newVolume.setInstanceId(null);
             newVolume.setPoolId(destDataStore.getId());
+            newVolume.setPoolType(destDataStore.getPoolType());
             newVolume.setState(Volume.State.Creating);
             newVolume.setPath(destData.getPath());
             newVolume.set_iScsiName(destData.getPath());

server/src/main/java/org/apache/cloudstack/storage/volume/VolumeImportUnmanageManagerImpl.java

@@ -452,7 +452,7 @@ public class VolumeImportUnmanageManagerImpl implements VolumeImportUnmanageServ
                                           Account owner, StoragePoolVO pool, String volumeName) {
         DiskProfile diskProfile = volumeManager.importVolume(Volume.Type.DATADISK, volumeName, diskOffering,
                 volume.getVirtualSize(), null, null, pool.getDataCenterId(), volume.getHypervisorType(), null, null,
-                owner, null, pool.getId(), volume.getPath(), null);
+                owner, null, pool.getId(), pool.getPoolType(), volume.getPath(), null);
         return volumeDao.findById(diskProfile.getVolumeId());
     }
 

server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java

@@ -820,7 +820,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         }
         diskProfile.setSize(copyRemoteVolumeAnswer.getSize());
         DiskProfile profile = volumeManager.updateImportedVolume(type, diskOffering, vm, template, deviceId,
-                storagePool.getId(), copyRemoteVolumeAnswer.getFilename(), chainInfo, diskProfile);
+                storagePool.getId(), storagePool.getPoolType(), copyRemoteVolumeAnswer.getFilename(), chainInfo, diskProfile);
 
         return new Pair<>(profile, storagePool);
     }
@@ -836,7 +836,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         StoragePool storagePool = storagePools.get(0);
 
         DiskProfile profile = volumeManager.updateImportedVolume(type, diskOffering, vm, template, deviceId,
-                storagePool.getId(), diskPath, null, diskProfile);
+                storagePool.getId(), storagePool.getPoolType(), diskPath, null, diskProfile);
 
         return new Pair<>(profile, storagePool);
     }
@@ -847,7 +847,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         StoragePool storagePool = primaryDataStoreDao.findById(poolId);
 
         DiskProfile profile = volumeManager.updateImportedVolume(type, diskOffering, vm, template, deviceId,
-                poolId, diskPath, null, diskProfile);
+                poolId, storagePool.getPoolType(), diskPath, null, diskProfile);
 
         return new Pair<>(profile, storagePool);
     }
@@ -866,7 +866,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         }
         StoragePool storagePool = getStoragePool(disk, zone, cluster, diskOffering);
         DiskProfile profile = volumeManager.importVolume(type, name, diskOffering, diskSize,
-                minIops, maxIops, vm.getDataCenterId(), vm.getHypervisorType(), vm, template, owner, deviceId, storagePool.getId(), path, chainInfo);
+                minIops, maxIops, vm.getDataCenterId(), vm.getHypervisorType(), vm, template, owner, deviceId, storagePool.getId(), storagePool.getPoolType(), path, chainInfo);
 
         return new Pair<DiskProfile, StoragePool>(profile, storagePool);
     }

server/src/test/java/org/apache/cloudstack/storage/volume/VolumeImportUnmanageManagerImplTest.java

@@ -277,7 +277,7 @@ public class VolumeImportUnmanageManagerImplTest {
         doNothing().when(volumeApiService).validateCustomDiskOfferingSizeRange(anyLong());
         doReturn(true).when(volumeApiService).doesStoragePoolSupportDiskOffering(any(), any());
         doReturn(diskProfile).when(volumeManager).importVolume(any(), anyString(), any(), eq(virtualSize), isNull(), isNull(), anyLong(),
-                any(), isNull(), isNull(), any(), isNull(), anyLong(), anyString(), isNull());
+                any(), isNull(), isNull(), any(), isNull(), anyLong(), any(), anyString(), isNull());
         when(diskProfile.getVolumeId()).thenReturn(volumeId);
         when(volumeDao.findById(volumeId)).thenReturn(volumeVO);
 

utils/src/test/java/org/apache/cloudstack/utils/reflectiontostringbuilderutils/ReflectionToStringBuilderUtilsTest.java

@@ -53,7 +53,7 @@ public class ReflectionToStringBuilderUtilsTest extends TestCase {
     private static final String DEFAULT_MULTIPLE_VALUES_SEPARATOR = ",";
 
     @Before
-    public void setup(){
+    public void setup() {
         classToReflect = String.class;
         classToReflectFieldsNamesList = ReflectionUtils.getAllFields(classToReflect).stream().map(objectField -> objectField.getName()).collect(Collectors.toList());
         classToReflectRemovedField = classToReflectFieldsNamesList.remove(0);