diff --git a/engine/components-api/src/main/java/com/cloud/template/TemplateManager.java b/engine/components-api/src/main/java/com/cloud/template/TemplateManager.java
index b8912526fdf2..a0f91882c4da 100644
--- a/engine/components-api/src/main/java/com/cloud/template/TemplateManager.java
+++ b/engine/components-api/src/main/java/com/cloud/template/TemplateManager.java
@@ -56,6 +56,13 @@ public interface TemplateManager {
             + "will validate if the provided URL is resolvable during the register of templates/ISOs before persisting them in the database.",
             true);
 
+    ConfigKey<Boolean> TemplateDeleteFromPrimaryStorage = new ConfigKey<Boolean>("Advanced",
+            Boolean.class,
+            "template.delete.from.primary.storage", "true",
+            "Template when deleted will be instantly deleted from the Primary Storage",
+            true,
+            ConfigKey.Scope.Global);
+
     static final String VMWARE_TOOLS_ISO = "vmware-tools.iso";
     static final String XS_TOOLS_ISO = "xs-tools.iso";
 
@@ -103,6 +110,8 @@ public interface TemplateManager {
      */
     List<VMTemplateStoragePoolVO> getUnusedTemplatesInPool(StoragePoolVO pool);
 
+    void evictTemplateFromStoragePoolsForZones(Long templateId, List<Long> zoneId);
+
     /**
      * Deletes a template in the specified storage pool.
      *
diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDao.java b/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDao.java
index a3ce03a74c34..a208590e23a3 100644
--- a/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDao.java
+++ b/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDao.java
@@ -35,6 +35,8 @@ public interface VMTemplatePoolDao extends GenericDao<VMTemplateStoragePoolVO, L
 
     List<VMTemplateStoragePoolVO> listByPoolIdAndState(long poolId, ObjectInDataStoreStateMachine.State state);
 
+    List<VMTemplateStoragePoolVO> listByPoolIdsAndTemplate(List<Long> poolIds, Long templateId);
+
     List<VMTemplateStoragePoolVO> listByTemplateStatus(long templateId, VMTemplateStoragePoolVO.Status downloadState);
 
     List<VMTemplateStoragePoolVO> listByTemplateStatus(long templateId, VMTemplateStoragePoolVO.Status downloadState, long poolId);
diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDaoImpl.java b/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDaoImpl.java
index 5a2ec1163fb0..5dfc138d8e1b 100644
--- a/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDaoImpl.java
+++ b/engine/schema/src/main/java/com/cloud/storage/dao/VMTemplatePoolDaoImpl.java
@@ -150,6 +150,16 @@ public VMTemplateStoragePoolVO findByPoolTemplate(long poolId, long templateId,
         return findOneIncludingRemovedBy(sc);
     }
 
+    @Override
+    public List<VMTemplateStoragePoolVO> listByPoolIdsAndTemplate(List<Long> poolIds, Long templateId) {
+        SearchCriteria<VMTemplateStoragePoolVO> sc = PoolTemplateSearch.create();
+        if (CollectionUtils.isNotEmpty(poolIds)) {
+            sc.setParameters("pool_id", poolIds.toArray());
+        }
+        sc.setParameters("template_id", templateId);
+        return listBy(sc);
+    }
+
     @Override
     public List<VMTemplateStoragePoolVO> listByTemplateStatus(long templateId, VMTemplateStoragePoolVO.Status downloadState) {
         SearchCriteria<VMTemplateStoragePoolVO> sc = TemplateStatusSearch.create();
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDao.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDao.java
index f0c235e842c1..102df6848f87 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDao.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDao.java
@@ -154,4 +154,6 @@ Pair<List<Long>, Integer> searchForIdsAndCount(Long storagePoolId, String storag
             String keyword, Filter searchFilter);
 
     List<StoragePoolVO> listByIds(List<Long> ids);
+
+    List<StoragePoolVO> listByDataCenterIds(List<Long> dataCenterIds);
 }
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDaoImpl.java
index cb7313954dc7..0d901c82306b 100644
--- a/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDaoImpl.java
+++ b/engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/PrimaryDataStoreDaoImpl.java
@@ -63,6 +63,7 @@ public class PrimaryDataStoreDaoImpl extends GenericDaoBase<StoragePoolVO, Long>
     private final GenericSearchBuilder<StoragePoolVO, Long> StatusCountSearch;
     private final SearchBuilder<StoragePoolVO> ClustersSearch;
     private final SearchBuilder<StoragePoolVO> IdsSearch;
+    private final SearchBuilder<StoragePoolVO> DcsSearch;
 
     @Inject
     private StoragePoolDetailsDao _detailsDao;
@@ -155,6 +156,9 @@ public PrimaryDataStoreDaoImpl() {
         IdsSearch.and("ids", IdsSearch.entity().getId(), SearchCriteria.Op.IN);
         IdsSearch.done();
 
+        DcsSearch = createSearchBuilder();
+        DcsSearch.and("dataCenterId", DcsSearch.entity().getDataCenterId(), SearchCriteria.Op.IN);
+        DcsSearch.done();
     }
 
     @Override
@@ -733,6 +737,16 @@ public List<StoragePoolVO> listByIds(List<Long> ids) {
         return listBy(sc);
     }
 
+    @Override
+    public List<StoragePoolVO> listByDataCenterIds(List<Long> dataCenterIds) {
+        if (CollectionUtils.isEmpty(dataCenterIds)) {
+            return Collections.emptyList();
+        }
+        SearchCriteria<StoragePoolVO> sc = DcsSearch.create();
+        sc.setParameters("dataCenterId", dataCenterIds.toArray());
+        return listBy(sc);
+    }
+
     private SearchCriteria<StoragePoolVO> createStoragePoolSearchCriteria(Long storagePoolId, String storagePoolName,
             Long zoneId, String path, Long podId, Long clusterId, String address, ScopeType scopeType,
             StoragePoolStatus status, String keyword) {
diff --git a/server/src/main/java/com/cloud/template/HypervisorTemplateAdapter.java b/server/src/main/java/com/cloud/template/HypervisorTemplateAdapter.java
index f7eb654141d4..8d38aba0e7e2 100644
--- a/server/src/main/java/com/cloud/template/HypervisorTemplateAdapter.java
+++ b/server/src/main/java/com/cloud/template/HypervisorTemplateAdapter.java
@@ -627,7 +627,7 @@ public boolean delete(TemplateProfile profile) {
 
                 boolean dataDiskDeletetionResult = true;
                 List<VMTemplateVO> dataDiskTemplates = templateDao.listByParentTemplatetId(template.getId());
-                if (dataDiskTemplates != null && dataDiskTemplates.size() > 0) {
+                if (CollectionUtils.isNotEmpty(dataDiskTemplates)) {
                     logger.info("Template: {} has Datadisk template(s) associated with it. Delete Datadisk templates before deleting the template", template);
                     for (VMTemplateVO dataDiskTemplate : dataDiskTemplates) {
                         logger.info("Delete Datadisk template: {} from image store: {}", dataDiskTemplate, imageStore);
@@ -693,6 +693,9 @@ public boolean delete(TemplateProfile profile) {
         if (success) {
             if ((imageStores != null && imageStores.size() > 1) && (profile.getZoneIdList() != null)) {
                 //if template is stored in more than one image stores, and the zone id is not null, then don't delete other templates.
+                if (templateMgr.TemplateDeleteFromPrimaryStorage.value()) {
+                    templateMgr.evictTemplateFromStoragePoolsForZones(template.getId(), profile.getZoneIdList());
+                }
                 return cleanupTemplate(template, success);
             }
 
@@ -705,7 +708,7 @@ public boolean delete(TemplateProfile profile) {
 
             // find all eligible image stores for this template
             List<DataStore> iStores = templateMgr.getImageStoreByTemplate(template.getId(), null);
-            if (iStores == null || iStores.size() == 0) {
+            if (CollectionUtils.isEmpty(iStores)) {
                 // remove any references from template_zone_ref
                 List<VMTemplateZoneVO> templateZones = templateZoneDao.listByTemplateId(template.getId());
                 if (templateZones != null) {
@@ -726,6 +729,10 @@ public boolean delete(TemplateProfile profile) {
 
             }
 
+            if (templateMgr.TemplateDeleteFromPrimaryStorage.value()) {
+                templateMgr.evictTemplateFromStoragePoolsForZones(template.getId(), profile.getZoneIdList());
+            }
+
             // remove its related ACL permission
             Pair<Class<?>, Long> templateClassForId = new Pair<>(VirtualMachineTemplate.class, template.getId());
             _messageBus.publish(_name, EntityManager.MESSAGE_REMOVE_ENTITY_EVENT, PublishScope.LOCAL, templateClassForId);
diff --git a/server/src/main/java/com/cloud/template/TemplateManagerImpl.java b/server/src/main/java/com/cloud/template/TemplateManagerImpl.java
index cf88ccec919c..5aa31eff1702 100755
--- a/server/src/main/java/com/cloud/template/TemplateManagerImpl.java
+++ b/server/src/main/java/com/cloud/template/TemplateManagerImpl.java
@@ -1024,33 +1024,53 @@ public boolean delete(long userId, long templateId, Long zoneId) {
         return adapter.delete(new TemplateProfile(userId, template, zoneId));
     }
 
+    private Boolean templateIsUnusedInPool(VMTemplateStoragePoolVO templatePoolVO) {
+        VMTemplateVO template = _tmpltDao.findByIdIncludingRemoved(templatePoolVO.getTemplateId());
+
+        // If this is a routing template, consider it in use
+        if (template.getTemplateType() == TemplateType.SYSTEM) {
+            return false;
+        }
+
+        // If the template is not yet downloaded to the pool, consider it in use
+        if (templatePoolVO.getDownloadState() != Status.DOWNLOADED) {
+            return false;
+        }
+
+        if (template.getFormat() == ImageFormat.ISO || _volumeDao.isAnyVolumeActivelyUsingTemplateOnPool(template.getId(), templatePoolVO.getPoolId())) {
+            return false;
+        }
+        return true;
+    }
+
     @Override
     public List<VMTemplateStoragePoolVO> getUnusedTemplatesInPool(StoragePoolVO pool) {
         List<VMTemplateStoragePoolVO> unusedTemplatesInPool = new ArrayList<VMTemplateStoragePoolVO>();
         List<VMTemplateStoragePoolVO> allTemplatesInPool = _tmpltPoolDao.listByPoolId(pool.getId());
 
         for (VMTemplateStoragePoolVO templatePoolVO : allTemplatesInPool) {
-            VMTemplateVO template = _tmpltDao.findByIdIncludingRemoved(templatePoolVO.getTemplateId());
-
-            // If this is a routing template, consider it in use
-            if (template.getTemplateType() == TemplateType.SYSTEM) {
-                continue;
-            }
-
-            // If the template is not yet downloaded to the pool, consider it in
-            // use
-            if (templatePoolVO.getDownloadState() != Status.DOWNLOADED) {
-                continue;
-            }
-
-            if (template.getFormat() != ImageFormat.ISO && !_volumeDao.isAnyVolumeActivelyUsingTemplateOnPool(template.getId(), pool.getId())) {
+            if (templateIsUnusedInPool(templatePoolVO)) {
                 unusedTemplatesInPool.add(templatePoolVO);
             }
         }
-
         return unusedTemplatesInPool;
     }
 
+    @Override
+    public void evictTemplateFromStoragePoolsForZones(Long templateId, List<Long> zoneIds) {
+        List<Long> poolIds = new ArrayList<>();
+        if (CollectionUtils.isNotEmpty(zoneIds)) {
+            List<StoragePoolVO> pools = _poolDao.listByDataCenterIds(zoneIds);
+            poolIds = pools.stream().map(StoragePoolVO::getId).collect(Collectors.toList());
+        }
+        List<VMTemplateStoragePoolVO> templateStoragePoolVOS = _tmpltPoolDao.listByPoolIdsAndTemplate(poolIds, templateId);
+        for (VMTemplateStoragePoolVO templateStoragePoolVO: templateStoragePoolVOS) {
+            if (templateIsUnusedInPool(templateStoragePoolVO)) {
+                evictTemplateFromStoragePool(templateStoragePoolVO);
+            }
+        }
+    }
+
     @Override
     @DB
     public void evictTemplateFromStoragePool(VMTemplateStoragePoolVO templatePoolVO) {
@@ -2368,7 +2388,10 @@ public String getConfigComponentName() {
 
     @Override
     public ConfigKey<?>[] getConfigKeys() {
-        return new ConfigKey<?>[] {AllowPublicUserTemplates, TemplatePreloaderPoolSize, ValidateUrlIsResolvableBeforeRegisteringTemplate};
+        return new ConfigKey<?>[] {AllowPublicUserTemplates,
+                TemplatePreloaderPoolSize,
+                ValidateUrlIsResolvableBeforeRegisteringTemplate,
+                TemplateDeleteFromPrimaryStorage};
     }
 
     public List<TemplateAdapter> getTemplateAdapters() {