volume: enforce storage pool disable threshold when creating a volume on an explicit pool

createVolume with an admin-specified storageid bypassed the normal
allocator path and thus skipped the pool's disable-threshold and
allocated-capacity checks (checkUsagedSpace / checkPoolforSpace).
createVolumeOnStoragePool now calls StorageManager.storagePoolHasEnoughSpace
before creating the volume, matching the check already done for
volume migration.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: weizhouapache

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

@@ -1127,6 +1127,14 @@ private VolumeVO createVolumeOnStoragePool(Long volumeId, Long storageId) throws
             throw new InvalidParameterValueException(String.format("Disk offering: %s is not compatible with the storage pool", diskOffering.getUuid()));
         }
 
+        HypervisorType hypervisorType = _volsDao.getHypervisorType(volume.getId());
+        DiskProfile diskProfile = new DiskProfile(volume, diskOffering, hypervisorType);
+        Pair<Volume, DiskProfile> volumeDiskProfilePair = new Pair<>(volume, diskProfile);
+        if (!storageMgr.storagePoolHasEnoughSpace(Collections.singletonList(volumeDiskProfilePair), storagePool)) {
+            throw new InvalidParameterValueException(String.format("Cannot create volume %s on storage pool %s as the pool does not have enough space " +
+                    "or has crossed the disable threshold.", volume.getUuid(), storagePool.getName()));
+        }
+
         DataStore dataStore = dataStoreMgr.getDataStore(storageId, DataStoreRole.Primary);
         VolumeInfo volumeInfo = volFactory.getVolume(volumeId, dataStore);
         AsyncCallFuture<VolumeApiResult> createVolumeFuture = volService.createVolumeAsync(volumeInfo, dataStore);

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

@@ -20,6 +20,7 @@
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -2896,4 +2897,98 @@ public void testResizeVolumeInternal_VMware_VMRunning_ShouldThrowStateGuardError
                     e.getMessage() != null && e.getMessage().contains("VM should be in"));
         }
     }
+
+    /**
+     * createVolumeOnStoragePool must reject the request when the target pool has crossed its
+     * storage capacity disable threshold, instead of silently creating the volume there.
+     */
+    @Test
+    public void testCreateVolumeOnStoragePool_DisableThresholdCrossed_ShouldThrow()
+            throws ExecutionException, InterruptedException {
+        long volumeId = 400L;
+        long storageId = 401L;
+        long diskOfferingId = 402L;
+        long dataCenterId = 1L;
+
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        when(volume.getId()).thenReturn(volumeId);
+        when(volume.getDataCenterId()).thenReturn(dataCenterId);
+        when(volume.getDiskOfferingId()).thenReturn(diskOfferingId);
+        when(volume.getUuid()).thenReturn("volume-uuid");
+        when(volumeDaoMock.findById(volumeId)).thenReturn(volume);
+        when(volumeDaoMock.getHypervisorType(volumeId)).thenReturn(HypervisorType.KVM);
+
+        PrimaryDataStore storagePool = Mockito.mock(PrimaryDataStore.class);
+        when(storagePool.getStatus()).thenReturn(StoragePoolStatus.Up);
+        when(storagePool.getDataCenterId()).thenReturn(dataCenterId);
+        when(storagePool.getName()).thenReturn("pool-crossing-threshold");
+        when(dataStoreMgr.getDataStore(storageId, DataStoreRole.Primary)).thenReturn(storagePool);
+
+        DiskOfferingVO diskOffering = Mockito.mock(DiskOfferingVO.class);
+        when(_diskOfferingDao.findById(diskOfferingId)).thenReturn(diskOffering);
+
+        Mockito.doReturn(true).when(volumeApiServiceImpl).doesStoragePoolSupportDiskOffering(storagePool, diskOffering);
+
+        // Simulate the pool having crossed its storage.capacity/allocated disable threshold.
+        when(storageMgr.storagePoolHasEnoughSpace(anyList(), eq(storagePool))).thenReturn(false);
+
+        try {
+            invokePrivateMethod("createVolumeOnStoragePool", new Class[]{Long.class, Long.class}, volumeId, storageId);
+            Assert.fail("Expected an InvalidParameterValueException because the pool has crossed its disable threshold");
+        } catch (RuntimeException e) {
+            // invokePrivateMethod wraps the reflectively-thrown exception as:
+            // RuntimeException -> InvocationTargetException -> actual exception
+            Throwable cause = e.getCause();
+            if (cause instanceof InvocationTargetException) {
+                cause = cause.getCause();
+            }
+            Assert.assertTrue("Expected InvalidParameterValueException, was: " + cause,
+                    cause instanceof InvalidParameterValueException);
+            Assert.assertTrue("Exception message must reference the disable threshold, was: " + cause.getMessage(),
+                    cause.getMessage() != null && cause.getMessage().contains("disable threshold"));
+        }
+
+        Mockito.verify(volumeServiceMock, Mockito.never()).createVolumeAsync(any(), any());
+    }
+
+    /**
+     * createVolumeOnStoragePool must proceed with volume creation when the target pool has
+     * enough space and has not crossed its disable threshold.
+     */
+    @Test
+    public void testCreateVolumeOnStoragePool_EnoughSpace_ShouldCreateVolume()
+            throws ExecutionException, InterruptedException {
+        long volumeId = 410L;
+        long storageId = 411L;
+        long diskOfferingId = 412L;
+        long dataCenterId = 1L;
+
+        VolumeVO volume = Mockito.mock(VolumeVO.class);
+        when(volume.getId()).thenReturn(volumeId);
+        when(volume.getDataCenterId()).thenReturn(dataCenterId);
+        when(volume.getDiskOfferingId()).thenReturn(diskOfferingId);
+        when(volumeDaoMock.findById(volumeId)).thenReturn(volume);
+        when(volumeDaoMock.getHypervisorType(volumeId)).thenReturn(HypervisorType.KVM);
+
+        PrimaryDataStore storagePool = Mockito.mock(PrimaryDataStore.class);
+        when(storagePool.getStatus()).thenReturn(StoragePoolStatus.Up);
+        when(storagePool.getDataCenterId()).thenReturn(dataCenterId);
+        when(dataStoreMgr.getDataStore(storageId, DataStoreRole.Primary)).thenReturn(storagePool);
+
+        DiskOfferingVO diskOffering = Mockito.mock(DiskOfferingVO.class);
+        when(_diskOfferingDao.findById(diskOfferingId)).thenReturn(diskOffering);
+
+        Mockito.doReturn(true).when(volumeApiServiceImpl).doesStoragePoolSupportDiskOffering(storagePool, diskOffering);
+        when(storageMgr.storagePoolHasEnoughSpace(anyList(), eq(storagePool))).thenReturn(true);
+
+        when(volumeDataFactoryMock.getVolume(volumeId, storagePool)).thenReturn(volumeInfoMock);
+        when(volumeInfoMock.getId()).thenReturn(volumeId);
+        when(volumeServiceMock.createVolumeAsync(volumeInfoMock, storagePool)).thenReturn(asyncCallFutureVolumeapiResultMock);
+
+        VolumeVO result = invokePrivateMethod("createVolumeOnStoragePool",
+                new Class[]{Long.class, Long.class}, volumeId, storageId);
+
+        Assert.assertEquals(volume, result);
+        Mockito.verify(volumeServiceMock).createVolumeAsync(volumeInfoMock, storagePool);
+    }
 }