diff --git a/engine/components-api/src/main/java/com/cloud/storage/StorageManager.java b/engine/components-api/src/main/java/com/cloud/storage/StorageManager.java
index 7976b313167..a62876aeff0 100644
--- a/engine/components-api/src/main/java/com/cloud/storage/StorageManager.java
+++ b/engine/components-api/src/main/java/com/cloud/storage/StorageManager.java
@@ -180,6 +180,8 @@ public interface StorageManager extends StorageService {
 
     public Answer getVolumeStats(StoragePool pool, Command cmd);
 
+    boolean canPoolProvideStorageStats(StoragePool pool);
+
     /**
      * Checks if a host has running VMs that are using its local storage pool.
      * @return true if local storage is active on the host
diff --git a/server/src/main/java/com/cloud/server/StatsCollector.java b/server/src/main/java/com/cloud/server/StatsCollector.java
index 36c0372325d..ce09ed0729d 100644
--- a/server/src/main/java/com/cloud/server/StatsCollector.java
+++ b/server/src/main/java/com/cloud/server/StatsCollector.java
@@ -1035,7 +1035,7 @@ public class StatsCollector extends ManagerBase implements ComponentMethodInterc
                                     s_logger.warn("Not setting capacity bytes, received " + ((StorageStats)answer).getCapacityBytes()  + " capacity for pool ID " + poolId);
                                 }
                             }
-                            if (pool.getUsedBytes() != ((StorageStats)answer).getByteUsed() && pool.getStorageProviderName().equalsIgnoreCase(DataStoreProvider.DEFAULT_PRIMARY)) {
+                            if (pool.getUsedBytes() != ((StorageStats)answer).getByteUsed() && (pool.getStorageProviderName().equalsIgnoreCase(DataStoreProvider.DEFAULT_PRIMARY) || _storageManager.canPoolProvideStorageStats(pool))) {
                                 pool.setUsedBytes(((StorageStats) answer).getByteUsed());
                                 poolNeedsUpdating = true;
                             }
@@ -1055,7 +1055,6 @@ public class StatsCollector extends ManagerBase implements ComponentMethodInterc
                 s_logger.error("Error trying to retrieve storage stats", t);
             }
         }
-
     }
 
     class AutoScaleMonitor extends ManagedContextRunnable {
diff --git a/server/src/main/java/com/cloud/storage/StorageManagerImpl.java b/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
index 6535843aed9..c340da29f40 100644
--- a/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
+++ b/server/src/main/java/com/cloud/storage/StorageManagerImpl.java
@@ -48,6 +48,7 @@ import javax.inject.Inject;
 import com.cloud.agent.api.GetStoragePoolCapabilitiesAnswer;
 import com.cloud.agent.api.GetStoragePoolCapabilitiesCommand;
 import com.cloud.network.router.VirtualNetworkApplianceManager;
+import com.cloud.server.StatsCollector;
 import com.cloud.upgrade.SystemVmTemplateRegistration;
 import org.apache.cloudstack.annotation.AnnotationService;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
@@ -469,13 +470,9 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
 
     @Override
     public Answer sendToPool(StoragePool pool, Command cmd) throws StorageUnavailableException {
-        if (cmd instanceof GetStorageStatsCommand) {
-            DataStoreProvider storeProvider = _dataStoreProviderMgr.getDataStoreProvider(pool.getStorageProviderName());
-            DataStoreDriver storeDriver = storeProvider.getDataStoreDriver();
-            if (storeDriver instanceof PrimaryDataStoreDriver && ((PrimaryDataStoreDriver)storeDriver).canProvideStorageStats()) {
-                // Get stats from the pool directly instead of sending cmd to host
-                return getStoragePoolStats(pool, (GetStorageStatsCommand) cmd);
-            }
+        if (cmd instanceof GetStorageStatsCommand && canPoolProvideStorageStats(pool)) {
+            // Get stats from the pool directly instead of sending cmd to host
+            return getStoragePoolStats(pool, (GetStorageStatsCommand) cmd);
         }
 
         Answer[] answers = sendToPool(pool, new Commands(cmd));
@@ -501,6 +498,13 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
         return answer;
     }
 
+    @Override
+    public boolean canPoolProvideStorageStats(StoragePool pool) {
+        DataStoreProvider storeProvider = _dataStoreProviderMgr.getDataStoreProvider(pool.getStorageProviderName());
+        DataStoreDriver storeDriver = storeProvider.getDataStoreDriver();
+        return storeDriver instanceof PrimaryDataStoreDriver && ((PrimaryDataStoreDriver)storeDriver).canProvideStorageStats();
+    }
+
     @Override
     public Answer getVolumeStats(StoragePool pool, Command cmd) {
         DataStoreProvider storeProvider = _dataStoreProviderMgr.getDataStoreProvider(pool.getStorageProviderName());
@@ -2265,14 +2269,14 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
 
     private boolean checkUsagedSpace(StoragePool pool) {
         // Managed storage does not currently deal with accounting for physically used space (only provisioned space). Just return true if "pool" is managed.
-        // StatsCollector gets the storage stats from the ScaleIO/PowerFlex pool directly, limit the usage based on the capacity disable threshold
-        if (pool.isManaged() && pool.getPoolType() != StoragePoolType.PowerFlex) {
+        if (pool.isManaged() && !canPoolProvideStorageStats(pool)) {
             return true;
         }
 
-        double storageUsedThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(pool.getDataCenterId());
         long totalSize = pool.getCapacityBytes();
-        double usedPercentage = ((double)pool.getUsedBytes() / (double)totalSize);
+        long usedSize = getUsedSize(pool);
+        double usedPercentage = ((double)usedSize / (double)totalSize);
+        double storageUsedThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(pool.getDataCenterId());
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("Checking pool " + pool.getId() + " for storage, totalSize: " + pool.getCapacityBytes() + ", usedBytes: " + pool.getUsedBytes() +
                     ", usedPct: " + usedPercentage + ", disable threshold: " + storageUsedThreshold);
@@ -2287,6 +2291,25 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
         return true;
     }
 
+    private long getUsedSize(StoragePool pool) {
+        if (pool.getStorageProviderName().equalsIgnoreCase(DataStoreProvider.DEFAULT_PRIMARY) || canPoolProvideStorageStats(pool)) {
+            return (pool.getUsedBytes());
+        }
+
+        StatsCollector sc = StatsCollector.getInstance();
+        if (sc != null) {
+            StorageStats stats = sc.getStoragePoolStats(pool.getId());
+            if (stats == null) {
+                stats = sc.getStorageStats(pool.getId());
+            }
+            if (stats != null) {
+                return (stats.getByteUsed());
+            }
+        }
+
+        return 0;
+    }
+
     @Override
     public boolean storagePoolHasEnoughIops(List<Volume> requestedVolumes, StoragePool pool) {
         if (requestedVolumes == null || requestedVolumes.isEmpty() || pool == null) {