Fix limitation on tag matching in 'migrateVolume' with disk offering replacement

When the feature to enable disk offering replacement during volume migration was created, we were forcing the tags of the new disk offering to exact the same as the tags of the target storage poll. However, that is not how ACS manages volumes allocation. This change modifies this validation to make it consistent with volume allocation.

Author: rafaelweingartner

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

@@ -19,6 +19,7 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -2114,9 +2115,9 @@ private DiskOfferingVO retrieveAndValidateNewDiskOffering(MigrateVolumeCmd cmd)
      * Performs the validations required for replacing the disk offering while migrating the volume of storage. If no new disk offering is provided, we do not execute any validation.
      * If a disk offering is informed, we then proceed with the following checks.
      * <ul>
-     *  <li>We check if the given volume is of ROOT type. We cannot change the disk offering of a ROOT volume. Therefore, we thrown an {@link InvalidParameterValueException}.
-     *  <li>We the disk is being migrated to shared storage and the new disk offering is for local storage (or vice versa), we throw an {@link InvalidParameterValueException}. Bear in mind that we are validating only the new disk offering. If none is provided we can override the current disk offering. This means, placing a volume with shared disk offering in local storage and vice versa.
-     *  <li>We then proceed checking if the tags of the new disk offerings match the tags of the target storage. If they do not match an {@link InvalidParameterValueException} is thrown.
+     *  <li>We check if the given volume is of ROOT type. We cannot change the disk offering of a ROOT volume. Therefore, we thrown an {@link InvalidParameterValueException};
+     *  <li>We the disk is being migrated to shared storage and the new disk offering is for local storage (or vice versa), we throw an {@link InvalidParameterValueException}. Bear in mind that we are validating only the new disk offering. If none is provided we can override the current disk offering. This means, placing a volume with shared disk offering in local storage and vice versa;
+     *  <li>We then proceed checking the target storage pool supports the new disk offering {@link #doesTargetStorageSupportNewDiskOffering(StoragePool, DiskOfferingVO)}.
      * </ul>
      *
      * 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.
@@ -2128,10 +2129,9 @@ protected void validateConditionsToReplaceDiskOfferingOfVolume(VolumeVO volume,
         if ((destPool.isShared() && newDiskOffering.getUseLocalStorage()) || destPool.isLocal() && newDiskOffering.isShared()) {
             throw new InvalidParameterValueException("You cannot move the volume to a shared storage and assing a disk offering for local storage and vice versa.");
         }
-        String storageTags = getStoragePoolTags(destPool);
-        if (!StringUtils.areTagsEqual(storageTags, newDiskOffering.getTags())) {
-            throw new InvalidParameterValueException(String.format("Target Storage [id=%s] tags [%s] does not match new disk offering [id=%s] tags [%s].", destPool.getUuid(), storageTags,
-                    newDiskOffering.getUuid(), newDiskOffering.getTags()));
+        if (!doesTargetStorageSupportNewDiskOffering(destPool, newDiskOffering)) {
+            throw new InvalidParameterValueException(String.format("Target Storage [id=%s] tags [%s] does not match new disk offering [id=%s] tags [%s].", destPool.getUuid(),
+                    getStoragePoolTags(destPool), newDiskOffering.getUuid(), newDiskOffering.getTags()));
         }
         if (volume.getSize() != newDiskOffering.getDiskSize()) {
             DiskOfferingVO oldDiskOffering = this._diskOfferingDao.findById(volume.getDiskOfferingId());
@@ -2142,6 +2142,54 @@ protected void validateConditionsToReplaceDiskOfferingOfVolume(VolumeVO volume,
         s_logger.info(String.format("Changing disk offering to [uuid=%s] while migrating volume [uuid=%s, name=%s].", newDiskOffering.getUuid(), volume.getUuid(), volume.getName()));
     }
 
+    /**
+     *  Checks if the target storage supports the new disk offering.
+     *  This validation is consistent with the mechanism used to select a storage pool to deploy a volume when a virtual machine is deployed or when a new data disk is allocated.
+     *
+     *  The scenarios when this method returns true or false is presented in the following table.
+     *
+     *   <table border="1">
+     *      <tr>
+     *          <th>#</th><th>Disk offering tags</th><th>Storage tags</th><th>Does the storage support the disk offering?</th>
+     *      </tr>
+     *      <body>
+     *      <tr>
+     *          <td>1</td><td>A,B</td><td>A</td><td>NO</td>
+     *      </tr>
+     *      <tr>
+     *          <td>2</td><td>A,B,C</td><td>A,B,C,D,X</td><td>YES</td>
+     *      </tr>
+     *      <tr>
+     *          <td>3</td><td>A,B,C</td><td>X,Y,Z</td><td>NO</td>
+     *      </tr>
+     *      <tr>
+     *          <td>4</td><td>null</td><td>A,S,D</td><td>YES</td>
+     *      </tr>
+     *      <tr>
+     *          <td>5</td><td>A</td><td>null</td><td>NO</td>
+     *      </tr>
+     *      <tr>
+     *          <td>6</td><td>null</td><td>null</td><td>YES</td>
+     *      </tr>
+     *      </body>
+     *   </table>
+     */
+    protected boolean doesTargetStorageSupportNewDiskOffering(StoragePool destPool, DiskOfferingVO newDiskOffering) {
+        String newDiskOfferingTags = newDiskOffering.getTags();
+        if (org.apache.commons.lang.StringUtils.isBlank(newDiskOfferingTags)) {
+            return true;
+        }
+        String storageTags = getStoragePoolTags(destPool);
+        boolean isStorageTagsBlank = org.apache.commons.lang.StringUtils.isBlank(storageTags);
+        if (isStorageTagsBlank) {
+            return false;
+        }
+        String[] storageTagsAsStringArray = org.apache.commons.lang.StringUtils.split(storageTags, ",");
+        String[] newDiskOfferingTagsAsStringArray = org.apache.commons.lang.StringUtils.split(newDiskOfferingTags, ",");
+
+        return CollectionUtils.isSubCollection(Arrays.asList(newDiskOfferingTagsAsStringArray), Arrays.asList(storageTagsAsStringArray));
+    }
+
     /**
      *  Retrieves the storage pool tags as a {@link String}. If the storage pool does not have tags we return a null value.
      */

server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java

@@ -580,9 +580,86 @@ public void validateConditionsToReplaceDiskOfferingOfVolumeTestEverythingWorking
         inOrder.verify(storagePoolMock).isLocal();
         inOrder.verify(newDiskOfferingMock, times(0)).isShared();
         inOrder.verify(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
-        inOrder.verify(newDiskOfferingMock).getTags();
 
         inOrder.verify(volumeVOMock).getSize();
         inOrder.verify(newDiskOfferingMock).getDiskSize();
     }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingMoreTagsThanStorageTags() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("A,B,C").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("A").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertFalse(result);
+    }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingTagsIsSubSetOfStorageTags() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("A,B,C").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("A,B,C,D,X,Y").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertTrue(result);
+    }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingTagsEmptyAndStorageTagsNotEmpty() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("A,B,C,D,X,Y").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertTrue(result);
+    }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingTagsNotEmptyAndStorageTagsEmpty() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("A").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertFalse(result);
+    }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingTagsEmptyAndStorageTagsEmpty() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertTrue(result);
+    }
+
+    @Test
+    public void doesTargetStorageSupportNewDiskOfferingTestDiskOfferingTagsDifferentFromdStorageTags() {
+        DiskOfferingVO diskOfferingVoMock = Mockito.mock(DiskOfferingVO.class);
+        Mockito.doReturn("A,B").when(diskOfferingVoMock).getTags();
+
+        StoragePool storagePoolMock = Mockito.mock(StoragePool.class);
+        Mockito.doReturn("C,D").when(volumeApiServiceImpl).getStoragePoolTags(storagePoolMock);
+
+        boolean result = volumeApiServiceImpl.doesTargetStorageSupportNewDiskOffering(storagePoolMock, diskOfferingVoMock);
+
+        Assert.assertFalse(result);
+    }
 }