Allow volume migrations in ScaleIO within and across ScaleIO storage clusters

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

@@ -212,6 +212,7 @@
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.ScopeType;
 import com.cloud.storage.Storage.ImageFormat;
+import com.cloud.storage.Storage;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.VMTemplateVO;
@@ -2904,7 +2905,7 @@ protected Map<Volume, StoragePool> buildMapUsingUserInformation(VirtualMachinePr
      *  </ul>
      */
     protected void executeManagedStorageChecksWhenTargetStoragePoolProvided(StoragePoolVO currentPool, VolumeVO volume, StoragePoolVO targetPool) {
-        if (!currentPool.isManaged()) {
+        if (!currentPool.isManaged() || currentPool.getPoolType().equals(Storage.StoragePoolType.PowerFlex)) {
             return;
         }
         if (currentPool.getId() == targetPool.getId()) {

engine/orchestration/src/test/java/com/cloud/vm/VirtualMachineManagerImplTest.java

@@ -32,6 +32,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
+import java.util.stream.Collectors;
 
 import org.apache.cloudstack.engine.subsystem.api.storage.StoragePoolAllocator;
 import org.apache.cloudstack.framework.config.ConfigKey;
@@ -46,6 +48,7 @@
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.Spy;
+import org.mockito.stubbing.Answer;
 import org.mockito.runners.MockitoJUnitRunner;
 
 import com.cloud.agent.AgentManager;
@@ -68,6 +71,7 @@
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.ScopeType;
+import com.cloud.storage.Storage;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.StoragePoolHostVO;
@@ -373,9 +377,26 @@ public void executeManagedStorageChecksWhenTargetStoragePoolProvidedTestCurrentS
         Mockito.verify(storagePoolVoMock, Mockito.times(0)).getId();
     }
 
+    @Test
+    public void allowVolumeMigrationsForPowerFlexStorage() {
+        Mockito.doReturn(true).when(storagePoolVoMock).isManaged();
+        Mockito.doReturn(Storage.StoragePoolType.PowerFlex).when(storagePoolVoMock).getPoolType();
+
+        virtualMachineManagerImpl.executeManagedStorageChecksWhenTargetStoragePoolProvided(storagePoolVoMock, volumeVoMock, Mockito.mock(StoragePoolVO.class));
+
+        Mockito.verify(storagePoolVoMock).isManaged();
+        Mockito.verify(storagePoolVoMock, Mockito.times(0)).getId();
+    }
+
     @Test
     public void executeManagedStorageChecksWhenTargetStoragePoolProvidedTestCurrentStoragePoolEqualsTargetPool() {
         Mockito.doReturn(true).when(storagePoolVoMock).isManaged();
+        // return any storage type except powerflex/scaleio
+        List<Storage.StoragePoolType> values = Arrays.asList(Storage.StoragePoolType.values());
+        when(storagePoolVoMock.getPoolType()).thenAnswer((Answer<Storage.StoragePoolType>) invocation -> {
+            List<Storage.StoragePoolType> filteredValues = values.stream().filter(v -> v != Storage.StoragePoolType.PowerFlex).collect(Collectors.toList());
+            int randomIndex = new Random().nextInt(filteredValues.size());
+            return filteredValues.get(randomIndex); });
 
         virtualMachineManagerImpl.executeManagedStorageChecksWhenTargetStoragePoolProvided(storagePoolVoMock, volumeVoMock, storagePoolVoMock);
 
@@ -386,6 +407,12 @@ public void executeManagedStorageChecksWhenTargetStoragePoolProvidedTestCurrentS
     @Test(expected = CloudRuntimeException.class)
     public void executeManagedStorageChecksWhenTargetStoragePoolProvidedTestCurrentStoragePoolNotEqualsTargetPool() {
         Mockito.doReturn(true).when(storagePoolVoMock).isManaged();
+        // return any storage type except powerflex/scaleio
+        List<Storage.StoragePoolType> values = Arrays.asList(Storage.StoragePoolType.values());
+        when(storagePoolVoMock.getPoolType()).thenAnswer((Answer<Storage.StoragePoolType>) invocation -> {
+            List<Storage.StoragePoolType> filteredValues = values.stream().filter(v -> v != Storage.StoragePoolType.PowerFlex).collect(Collectors.toList());
+            int randomIndex = new Random().nextInt(filteredValues.size());
+            return filteredValues.get(randomIndex); });
 
         virtualMachineManagerImpl.executeManagedStorageChecksWhenTargetStoragePoolProvided(storagePoolVoMock, volumeVoMock, Mockito.mock(StoragePoolVO.class));
     }

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

@@ -1648,6 +1648,10 @@ protected VolumeVO duplicateVolumeOnAnotherStorage(Volume volume, StoragePool po
         newVol.setPoolType(pool.getPoolType());
         newVol.setLastPoolId(lastPoolId);
         newVol.setPodId(pool.getPodId());
+        if (volume.getPassphraseId() != null) {
+            newVol.setPassphraseId(volume.getPassphraseId());
+            newVol.setEncryptFormat(volume.getEncryptFormat());
+        }
         return volDao.persist(newVol);
     }
 

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtConnection.java

@@ -21,6 +21,7 @@
 
 import org.apache.log4j.Logger;
 import org.libvirt.Connect;
+import org.libvirt.Library;
 import org.libvirt.LibvirtException;
 
 import com.cloud.hypervisor.Hypervisor;
@@ -44,6 +45,7 @@ static public Connect getConnection(String hypervisorURI) throws LibvirtExceptio
         if (conn == null) {
             s_logger.info("No existing libvirtd connection found. Opening a new one");
             conn = new Connect(hypervisorURI, false);
+            Library.initEventLoop();
             s_logger.debug("Successfully connected to libvirt at: " + hypervisorURI);
             s_connections.put(hypervisorURI, conn);
         } else {

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtMigrateVolumeCommandWrapper.java

@@ -29,27 +29,255 @@
 import com.cloud.hypervisor.kvm.storage.KVMStoragePoolManager;
 import com.cloud.resource.CommandWrapper;
 import com.cloud.resource.ResourceWrapper;
+import com.cloud.storage.Storage;
 
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.StringWriter;
 import java.util.Map;
 import java.util.UUID;
 
+import org.apache.cloudstack.storage.datastore.client.ScaleIOGatewayClient;
+import org.apache.cloudstack.storage.datastore.util.ScaleIOUtil;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 import org.apache.cloudstack.storage.to.VolumeObjectTO;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
+import org.libvirt.Connect;
+import org.libvirt.Domain;
+import org.libvirt.DomainBlockJobInfo;
+import org.libvirt.DomainInfo;
+import org.libvirt.TypedParameter;
+import org.libvirt.TypedUlongParameter;
+import org.libvirt.LibvirtException;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
+import org.xml.sax.SAXException;
+
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.transform.OutputKeys;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerException;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.transform.stream.StreamResult;
 
 @ResourceWrapper(handles =  MigrateVolumeCommand.class)
-public final class LibvirtMigrateVolumeCommandWrapper extends CommandWrapper<MigrateVolumeCommand, Answer, LibvirtComputingResource> {
+public class LibvirtMigrateVolumeCommandWrapper extends CommandWrapper<MigrateVolumeCommand, Answer, LibvirtComputingResource> {
     private static final Logger LOGGER = Logger.getLogger(LibvirtMigrateVolumeCommandWrapper.class);
 
     @Override
     public Answer execute(final MigrateVolumeCommand command, final LibvirtComputingResource libvirtComputingResource) {
+        VolumeObjectTO srcVolumeObjectTO = (VolumeObjectTO)command.getSrcData();
+        PrimaryDataStoreTO srcPrimaryDataStore = (PrimaryDataStoreTO)srcVolumeObjectTO.getDataStore();
+
+        MigrateVolumeAnswer answer;
+        if (srcPrimaryDataStore.getPoolType().equals(Storage.StoragePoolType.PowerFlex)) {
+            answer = migratePowerFlexVolume(command, libvirtComputingResource);
+        } else {
+            answer = migrateRegularVolume(command, libvirtComputingResource);
+        }
+
+        return answer;
+    }
+
+    protected MigrateVolumeAnswer migratePowerFlexVolume(final MigrateVolumeCommand command, final LibvirtComputingResource libvirtComputingResource) {
+
+        // Source Details
+        VolumeObjectTO srcVolumeObjectTO = (VolumeObjectTO)command.getSrcData();
+        String srcPath = srcVolumeObjectTO.getPath();
+        final String srcVolumeId = ScaleIOUtil.getVolumePath(srcVolumeObjectTO.getPath());
+        final String vmName = srcVolumeObjectTO.getVmName();
+
+        // Destination Details
+        VolumeObjectTO destVolumeObjectTO = (VolumeObjectTO)command.getDestData();
+        String destPath = destVolumeObjectTO.getPath();
+        final String destVolumeId = ScaleIOUtil.getVolumePath(destVolumeObjectTO.getPath());
+        Map<String, String> destDetails = command.getDestDetails();
+        final String destSystemId = destDetails.get(ScaleIOGatewayClient.STORAGE_POOL_SYSTEM_ID);
+        String destDiskLabel = null;
+
+        final String destDiskFileName = ScaleIOUtil.DISK_NAME_PREFIX + destSystemId + "-" + destVolumeId;
+        final String diskFilePath = ScaleIOUtil.DISK_PATH + File.separator + destDiskFileName;
+
+        Domain dm = null;
+        try {
+            final LibvirtUtilitiesHelper libvirtUtilitiesHelper = libvirtComputingResource.getLibvirtUtilitiesHelper();
+            Connect conn = libvirtUtilitiesHelper.getConnection();
+            dm = libvirtComputingResource.getDomain(conn, vmName);
+            if (dm == null) {
+                return new MigrateVolumeAnswer(command, false, "Migrate volume failed due to can not find vm: " + vmName, null);
+            }
+
+            DomainInfo.DomainState domainState = dm.getInfo().state ;
+            if (domainState != DomainInfo.DomainState.VIR_DOMAIN_RUNNING) {
+                return new MigrateVolumeAnswer(command, false, "Migrate volume failed due to VM is not running: " + vmName + " with domainState = " + domainState, null);
+            }
+
+            final KVMStoragePoolManager storagePoolMgr = libvirtComputingResource.getStoragePoolMgr();
+            PrimaryDataStoreTO spool = (PrimaryDataStoreTO)destVolumeObjectTO.getDataStore();
+            KVMStoragePool pool = storagePoolMgr.getStoragePool(spool.getPoolType(), spool.getUuid());
+            pool.connectPhysicalDisk(destVolumeObjectTO.getPath(), null);
+
+            String srcSecretUUID = null;
+            String destSecretUUID = null;
+            if (ArrayUtils.isNotEmpty(destVolumeObjectTO.getPassphrase())) {
+                srcSecretUUID = libvirtComputingResource.createLibvirtVolumeSecret(conn, srcVolumeObjectTO.getPath(), srcVolumeObjectTO.getPassphrase());
+                destSecretUUID = libvirtComputingResource.createLibvirtVolumeSecret(conn, destVolumeObjectTO.getPath(), destVolumeObjectTO.getPassphrase());
+            }
+
+            String diskdef = generateDestinationDiskXML(dm, srcVolumeId, diskFilePath, destSecretUUID);
+            destDiskLabel = generateDestinationDiskLabel(diskdef);
+
+            TypedUlongParameter parameter = new TypedUlongParameter("bandwidth", 0);
+            TypedParameter[] parameters = new TypedParameter[1];
+            parameters[0] = parameter;
+
+            dm.blockCopy(destDiskLabel, diskdef, parameters, Domain.BlockCopyFlags.REUSE_EXT);
+            LOGGER.info(String.format("Block copy has started for the volume %s : %s ", destDiskLabel, srcPath));
+
+            return checkBlockJobStatus(command, dm, destDiskLabel, srcPath, destPath, libvirtComputingResource, conn, srcSecretUUID);
+
+        } catch (Exception e) {
+            String msg = "Migrate volume failed due to " + e.toString();
+            LOGGER.warn(msg, e);
+            if (destDiskLabel != null) {
+                try {
+                    dm.blockJobAbort(destDiskLabel, Domain.BlockJobAbortFlags.ASYNC);
+                } catch (LibvirtException ex) {
+                    LOGGER.error("Migrate volume failed while aborting the block job due to " + ex.getMessage());
+                }
+            }
+            return new MigrateVolumeAnswer(command, false, msg, null);
+        } finally {
+            if (dm != null) {
+                try {
+                    dm.free();
+                } catch (LibvirtException l) {
+                    LOGGER.trace("Ignoring libvirt error.", l);
+                };
+            }
+        }
+    }
+
+    protected MigrateVolumeAnswer checkBlockJobStatus(MigrateVolumeCommand command, Domain dm, String diskLabel, String srcPath, String destPath, LibvirtComputingResource libvirtComputingResource, Connect conn, String srcSecretUUID) throws LibvirtException {
+        int timeBetweenTries = 1000; // Try more frequently (every sec) and return early if disk is found
+        int waitTimeInSec = command.getWait();
+        while (waitTimeInSec > 0) {
+            DomainBlockJobInfo blockJobInfo = dm.getBlockJobInfo(diskLabel, 0);
+            if (blockJobInfo != null) {
+                LOGGER.debug(String.format("Volume %s : %s block copy progress: %s%% current value:%s end value:%s", diskLabel, srcPath, (blockJobInfo.end == 0)? 0 : 100*(blockJobInfo.cur / (double) blockJobInfo.end), blockJobInfo.cur, blockJobInfo.end));
+                if (blockJobInfo.cur == blockJobInfo.end) {
+                    LOGGER.info(String.format("Block copy completed for the volume %s : %s", diskLabel, srcPath));
+                    dm.blockJobAbort(diskLabel, Domain.BlockJobAbortFlags.PIVOT);
+                    if (StringUtils.isNotEmpty(srcSecretUUID)) {
+                        libvirtComputingResource.removeLibvirtVolumeSecret(conn, srcSecretUUID);
+                    }
+                    break;
+                }
+            } else {
+                LOGGER.info("Failed to get the block copy status, trying to abort the job");
+                dm.blockJobAbort(diskLabel, Domain.BlockJobAbortFlags.ASYNC);
+            }
+            waitTimeInSec--;
+
+            try {
+                Thread.sleep(timeBetweenTries);
+            } catch (Exception ex) {
+                // don't do anything
+            }
+        }
+
+        if (waitTimeInSec <= 0) {
+            String msg = "Block copy is taking long time, failing the job";
+            LOGGER.error(msg);
+            try {
+                dm.blockJobAbort(diskLabel, Domain.BlockJobAbortFlags.ASYNC);
+            } catch (LibvirtException ex) {
+                LOGGER.error("Migrate volume failed while aborting the block job due to " + ex.getMessage());
+            }
+            return new MigrateVolumeAnswer(command, false, msg, null);
+        }
+
+        return new MigrateVolumeAnswer(command, true, null, destPath);
+    }
+
+    private String generateDestinationDiskLabel(String diskXml) throws ParserConfigurationException, IOException, SAXException {
+
+        DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
+        DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();
+        Document doc = dBuilder.parse(new ByteArrayInputStream(diskXml.getBytes("UTF-8")));
+        doc.getDocumentElement().normalize();
+
+        Element disk = doc.getDocumentElement();
+        String diskLabel = getAttrValue("target", "dev", disk);
+
+        return diskLabel;
+    }
+
+    protected String generateDestinationDiskXML(Domain dm, String srcVolumeId, String diskFilePath, String destSecretUUID) throws LibvirtException, ParserConfigurationException, IOException, TransformerException, SAXException {
+        final String domXml = dm.getXMLDesc(0);
+
+        DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
+        DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();
+        Document doc = dBuilder.parse(new ByteArrayInputStream(domXml.getBytes("UTF-8")));
+        doc.getDocumentElement().normalize();
+
+        NodeList disks = doc.getElementsByTagName("disk");
+
+        for (int i = 0; i < disks.getLength(); i++) {
+            Element disk = (Element)disks.item(i);
+            String type = disk.getAttribute("type");
+            if (!type.equalsIgnoreCase("network")) {
+                String diskDev = getAttrValue("source", "dev", disk);
+                if (StringUtils.isNotEmpty(diskDev) && diskDev.contains(srcVolumeId)) {
+                    setAttrValue("source", "dev", diskFilePath, disk);
+                    if (StringUtils.isNotEmpty(destSecretUUID)) {
+                        setAttrValue("secret", "uuid", destSecretUUID, disk);
+                    }
+                    StringWriter diskSection = new StringWriter();
+                    Transformer xformer = TransformerFactory.newInstance().newTransformer();
+                    xformer.setOutputProperty(OutputKeys.OMIT_XML_DECLARATION, "yes");
+                    xformer.transform(new DOMSource(disk), new StreamResult(diskSection));
+
+                    return diskSection.toString();
+                }
+            }
+        }
+
+        return null;
+    }
+
+    private static String getAttrValue(String tag, String attr, Element eElement) {
+        NodeList tagNode = eElement.getElementsByTagName(tag);
+        if (tagNode.getLength() == 0) {
+            return null;
+        }
+        Element node = (Element)tagNode.item(0);
+        return node.getAttribute(attr);
+    }
+
+    private static void setAttrValue(String tag, String attr, String newValue, Element eElement) {
+        NodeList tagNode = eElement.getElementsByTagName(tag);
+        if (tagNode.getLength() == 0) {
+            return;
+        }
+        Element node = (Element)tagNode.item(0);
+        node.setAttribute(attr, newValue);
+    }
+
+    protected MigrateVolumeAnswer migrateRegularVolume(final MigrateVolumeCommand command, final LibvirtComputingResource libvirtComputingResource) {
         KVMStoragePoolManager storagePoolManager = libvirtComputingResource.getStoragePoolMgr();
 
         VolumeObjectTO srcVolumeObjectTO = (VolumeObjectTO)command.getSrcData();
         PrimaryDataStoreTO srcPrimaryDataStore = (PrimaryDataStoreTO)srcVolumeObjectTO.getDataStore();
 
         Map<String, String> srcDetails = command.getSrcDetails();
-
         String srcPath = srcDetails != null ? srcDetails.get(DiskTO.IQN) : srcVolumeObjectTO.getPath();
 
         VolumeObjectTO destVolumeObjectTO = (VolumeObjectTO)command.getDestData();