diff --git a/api/src/main/java/org/apache/cloudstack/api/command/user/iso/RegisterIsoCmd.java b/api/src/main/java/org/apache/cloudstack/api/command/user/iso/RegisterIsoCmd.java
index ecab3930e8b..1d750038042 100644
--- a/api/src/main/java/org/apache/cloudstack/api/command/user/iso/RegisterIsoCmd.java
+++ b/api/src/main/java/org/apache/cloudstack/api/command/user/iso/RegisterIsoCmd.java
@@ -177,6 +177,9 @@ public class RegisterIsoCmd extends BaseCmd implements UserCmd {
     }
 
     public Long getZoneId() {
+        if (zoneId == null || zoneId == -1) {
+            return null;
+        }
         return zoneId;
     }
 
@@ -220,6 +223,10 @@ public class RegisterIsoCmd extends BaseCmd implements UserCmd {
         return directDownload == null ? false : directDownload;
     }
 
+    public void setDirectDownload(Boolean directDownload) {
+        this.directDownload = directDownload;
+    }
+
     public boolean isPasswordEnabled() {
         return passwordEnabled == null ? false : passwordEnabled;
     }
diff --git a/engine/storage/image/src/main/java/org/apache/cloudstack/storage/image/TemplateDataFactoryImpl.java b/engine/storage/image/src/main/java/org/apache/cloudstack/storage/image/TemplateDataFactoryImpl.java
index 492ec74382b..8951b9d7c24 100644
--- a/engine/storage/image/src/main/java/org/apache/cloudstack/storage/image/TemplateDataFactoryImpl.java
+++ b/engine/storage/image/src/main/java/org/apache/cloudstack/storage/image/TemplateDataFactoryImpl.java
@@ -20,11 +20,10 @@ package org.apache.cloudstack.storage.image;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import javax.inject.Inject;
 
-import com.cloud.hypervisor.Hypervisor;
-import com.cloud.utils.exception.CloudRuntimeException;
 import org.apache.cloudstack.direct.download.DirectDownloadManager;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
@@ -36,18 +35,21 @@ import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreVO;
 import org.apache.cloudstack.storage.image.store.TemplateObject;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Component;
 
 import com.cloud.host.HostVO;
 import com.cloud.host.dao.HostDao;
+import com.cloud.hypervisor.Hypervisor;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.VMTemplateStoragePoolVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VMTemplatePoolDao;
+import com.cloud.utils.exception.CloudRuntimeException;
 
 @Component
 public class TemplateDataFactoryImpl implements TemplateDataFactory {
@@ -203,12 +205,7 @@ public class TemplateDataFactoryImpl implements TemplateDataFactory {
      * Given existing spool refs, return one pool id existing on pools and refs
      */
     private Long getOneMatchingPoolIdFromRefs(List<VMTemplateStoragePoolVO> existingRefs, List<StoragePoolVO> pools) {
-        if (pools.isEmpty()) {
-            throw new CloudRuntimeException("No storage pools found");
-        }
-        if (existingRefs.isEmpty()) {
-            return pools.get(0).getId();
-        } else {
+        if (!existingRefs.isEmpty()) {
             for (VMTemplateStoragePoolVO ref : existingRefs) {
                 for (StoragePoolVO p : pools) {
                     if (ref.getPoolId() == p.getId()) {
@@ -217,45 +214,51 @@ public class TemplateDataFactoryImpl implements TemplateDataFactory {
                 }
             }
         }
-        return null;
+        return pools.get(0).getId();
     }
 
     /**
-     * Retrieve storage pools with scope = cluster or zone matching clusterId or dataCenterId depending on their scope
+     * Retrieve storage pools with scope = cluster or zone or local matching clusterId or dataCenterId or hostId depending on their scope
      */
-    private List<StoragePoolVO> getStoragePoolsFromClusterOrZone(Long clusterId, long dataCenterId, Hypervisor.HypervisorType hypervisorType) {
+    private List<StoragePoolVO> getStoragePoolsForScope(long dataCenterId, Long clusterId, long hostId, Hypervisor.HypervisorType hypervisorType) {
         List<StoragePoolVO> pools = new ArrayList<>();
         if (clusterId != null) {
             List<StoragePoolVO> clusterPools = primaryDataStoreDao.listPoolsByCluster(clusterId);
+            clusterPools = clusterPools.stream().filter(p -> !p.isLocal()).collect(Collectors.toList());
             pools.addAll(clusterPools);
         }
         List<StoragePoolVO> zonePools = primaryDataStoreDao.findZoneWideStoragePoolsByHypervisor(dataCenterId, hypervisorType);
         pools.addAll(zonePools);
+        List<StoragePoolVO> localPools = primaryDataStoreDao.findLocalStoragePoolsByHostAndTags(hostId, null);
+        pools.addAll(localPools);
         return pools;
     }
 
+    protected Long getBypassedTemplateExistingOrNewPoolId(VMTemplateVO templateVO, Long hostId) {
+        HostVO host = hostDao.findById(hostId);
+        List<StoragePoolVO> pools = getStoragePoolsForScope(host.getDataCenterId(), host.getClusterId(), hostId, host.getHypervisorType());
+        if (CollectionUtils.isEmpty(pools)) {
+            throw new CloudRuntimeException(String.format("No storage pool found to download template: %s", templateVO.getName()));
+        }
+        List<VMTemplateStoragePoolVO> existingRefs = templatePoolDao.listByTemplateId(templateVO.getId());
+        return getOneMatchingPoolIdFromRefs(existingRefs, pools);
+    }
+
     @Override
     public TemplateInfo getReadyBypassedTemplateOnPrimaryStore(long templateId, Long poolId, Long hostId) {
         VMTemplateVO templateVO = imageDataDao.findById(templateId);
         if (templateVO == null || !templateVO.isDirectDownload()) {
             return null;
         }
-        Long pool = poolId;
+        Long templatePoolId = poolId;
         if (poolId == null) {
-            //Get ISO from existing pool ref
-            HostVO host = hostDao.findById(hostId);
-            List<StoragePoolVO> pools = getStoragePoolsFromClusterOrZone(host.getClusterId(), host.getDataCenterId(), host.getHypervisorType());
-            List<VMTemplateStoragePoolVO> existingRefs = templatePoolDao.listByTemplateId(templateId);
-            pool = getOneMatchingPoolIdFromRefs(existingRefs, pools);
+            templatePoolId = getBypassedTemplateExistingOrNewPoolId(templateVO, hostId);
         }
-        if (pool == null) {
-            throw new CloudRuntimeException("No storage pool found where to download template: " + templateId);
-        }
-        VMTemplateStoragePoolVO spoolRef = templatePoolDao.findByPoolTemplate(pool, templateId, null);
+        VMTemplateStoragePoolVO spoolRef = templatePoolDao.findByPoolTemplate(templatePoolId, templateId, null);
         if (spoolRef == null) {
-            directDownloadManager.downloadTemplate(templateId, pool, hostId);
+            directDownloadManager.downloadTemplate(templateId, templatePoolId, hostId);
         }
-        DataStore store = storeMgr.getDataStore(pool, DataStoreRole.Primary);
+        DataStore store = storeMgr.getDataStore(templatePoolId, DataStoreRole.Primary);
         return this.getTemplate(templateId, store);
     }
 
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStoragePoolManager.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStoragePoolManager.java
index 987825921b8..b1842f38da2 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStoragePoolManager.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStoragePoolManager.java
@@ -294,12 +294,14 @@ public class KVMStoragePoolManager {
         String uuid = null;
         String sourceHost = "";
         StoragePoolType protocol = null;
-        if (storageUri.getScheme().equalsIgnoreCase("nfs") || storageUri.getScheme().equalsIgnoreCase("NetworkFilesystem")) {
+        final String scheme = storageUri.getScheme().toLowerCase();
+        List<String> acceptedSchemes = List.of("nfs", "networkfilesystem", "filesystem");
+        if (acceptedSchemes.contains(scheme)) {
             sourcePath = storageUri.getPath();
             sourcePath = sourcePath.replace("//", "/");
             sourceHost = storageUri.getHost();
             uuid = UUID.nameUUIDFromBytes(new String(sourceHost + sourcePath).getBytes()).toString();
-            protocol = StoragePoolType.NetworkFilesystem;
+            protocol = scheme.equals("filesystem") ? StoragePoolType.Filesystem: StoragePoolType.NetworkFilesystem;
         }
 
         // secondary storage registers itself through here
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
index f7ec09ca50f..dd31025d35f 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
@@ -25,23 +25,26 @@ import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 import javax.naming.ConfigurationException;
 
-import org.apache.cloudstack.direct.download.DirectDownloadHelper;
-import org.apache.cloudstack.direct.download.DirectTemplateDownloader;
-import com.cloud.storage.ScopeType;
-import com.cloud.storage.Volume;
 import org.apache.cloudstack.agent.directdownload.DirectDownloadAnswer;
 import org.apache.cloudstack.agent.directdownload.DirectDownloadCommand;
+import org.apache.cloudstack.direct.download.DirectDownloadHelper;
+import org.apache.cloudstack.direct.download.DirectTemplateDownloader;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
 import org.apache.cloudstack.storage.command.AttachAnswer;
 import org.apache.cloudstack.storage.command.AttachCommand;
@@ -71,7 +74,10 @@ import org.apache.cloudstack.utils.qemu.QemuImgFile;
 import org.apache.cloudstack.utils.qemu.QemuObject;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.io.FileUtils;
-
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 import org.apache.log4j.Logger;
 import org.libvirt.Connect;
 import org.libvirt.Domain;
@@ -110,9 +116,11 @@ import com.cloud.hypervisor.kvm.resource.LibvirtVMDef.DiskDef.DiskProtocol;
 import com.cloud.hypervisor.kvm.resource.wrapper.LibvirtUtilitiesHelper;
 import com.cloud.storage.JavaStorageLayer;
 import com.cloud.storage.MigrationOptions;
+import com.cloud.storage.ScopeType;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.StorageLayer;
+import com.cloud.storage.Volume;
 import com.cloud.storage.resource.StorageProcessor;
 import com.cloud.storage.template.Processor;
 import com.cloud.storage.template.Processor.FormatInfo;
@@ -126,16 +134,6 @@ import com.cloud.utils.script.Script;
 import com.cloud.utils.storage.S3.S3Utils;
 import com.cloud.vm.VmDetailConstants;
 
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.stream.Collectors;
-import org.apache.commons.lang3.BooleanUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.builder.ToStringBuilder;
-import org.apache.commons.lang3.builder.ToStringStyle;
-
 public class KVMStorageProcessor implements StorageProcessor {
     private static final Logger s_logger = Logger.getLogger(KVMStorageProcessor.class);
     private final KVMStoragePoolManager storagePoolMgr;
@@ -1074,7 +1072,8 @@ public class KVMStorageProcessor implements StorageProcessor {
             s_logger.debug(String.format("This backup is temporary, not deleting snapshot [%s] on primary storage [%s]", snapshotPath, primaryPool.getUuid()));
         }
     }
-    protected synchronized void attachOrDetachISO(final Connect conn, final String vmName, String isoPath, final boolean isAttach, Map<String, String> params) throws
+
+    protected synchronized void attachOrDetachISO(final Connect conn, final String vmName, String isoPath, final boolean isAttach, Map<String, String> params, DataStoreTO store) throws
             LibvirtException, InternalErrorException {
         DiskDef iso = new DiskDef();
         boolean isUefiEnabled = MapUtils.isNotEmpty(params) && params.containsKey("UEFI");
@@ -1082,8 +1081,14 @@ public class KVMStorageProcessor implements StorageProcessor {
             final int index = isoPath.lastIndexOf("/");
             final String path = isoPath.substring(0, index);
             final String name = isoPath.substring(index + 1);
-            final KVMStoragePool secondaryPool = storagePoolMgr.getStoragePoolByURI(path);
-            final KVMPhysicalDisk isoVol = secondaryPool.getPhysicalDisk(name);
+            KVMStoragePool storagePool;
+            if (store instanceof PrimaryDataStoreTO) {
+                PrimaryDataStoreTO primaryDataStoreTO = (PrimaryDataStoreTO)store;
+                storagePool = storagePoolMgr.getStoragePool(primaryDataStoreTO.getPoolType(), store.getUuid());
+            } else {
+                storagePool = storagePoolMgr.getStoragePoolByURI(path);
+            }
+            final KVMPhysicalDisk isoVol = storagePool.getPhysicalDisk(name);
             isoPath = isoVol.getPath();
 
             iso.defISODisk(isoPath, isUefiEnabled);
@@ -1112,7 +1117,7 @@ public class KVMStorageProcessor implements StorageProcessor {
         try {
             String dataStoreUrl = getDataStoreUrlFromStore(store);
             final Connect conn = LibvirtConnection.getConnectionByVmName(cmd.getVmName());
-            attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), true, cmd.getControllerInfo());
+            attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), true, cmd.getControllerInfo(), store);
         } catch (final LibvirtException e) {
             return new Answer(cmd, false, e.toString());
         } catch (final InternalErrorException e) {
@@ -1133,7 +1138,7 @@ public class KVMStorageProcessor implements StorageProcessor {
         try {
             String dataStoreUrl = getDataStoreUrlFromStore(store);
             final Connect conn = LibvirtConnection.getConnectionByVmName(cmd.getVmName());
-            attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), false, cmd.getParams());
+            attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), false, cmd.getParams(), store);
         } catch (final LibvirtException e) {
             return new Answer(cmd, false, e.toString());
         } catch (final InternalErrorException e) {
@@ -1149,19 +1154,25 @@ public class KVMStorageProcessor implements StorageProcessor {
      * Return data store URL from store
      */
     private String getDataStoreUrlFromStore(DataStoreTO store) {
-        if (!(store instanceof NfsTO) && (!(store instanceof PrimaryDataStoreTO) ||
-                store instanceof PrimaryDataStoreTO && !((PrimaryDataStoreTO) store).getPoolType().equals(StoragePoolType.NetworkFilesystem))) {
+        List<StoragePoolType> supportedPoolType = List.of(StoragePoolType.NetworkFilesystem, StoragePoolType.Filesystem);
+        if (!(store instanceof NfsTO) && (!(store instanceof PrimaryDataStoreTO) || !supportedPoolType.contains(((PrimaryDataStoreTO) store).getPoolType()))) {
+            s_logger.error(String.format("Unsupported protocol, store: %s", store.getUuid()));
             throw new InvalidParameterValueException("unsupported protocol");
         }
 
         if (store instanceof NfsTO) {
             NfsTO nfsStore = (NfsTO)store;
             return nfsStore.getUrl();
-        } else if (store instanceof PrimaryDataStoreTO && ((PrimaryDataStoreTO) store).getPoolType().equals(StoragePoolType.NetworkFilesystem)) {
+        } else if (store instanceof PrimaryDataStoreTO) {
             //In order to support directly downloaded ISOs
+            StoragePoolType poolType = ((PrimaryDataStoreTO)store).getPoolType();
             String psHost = ((PrimaryDataStoreTO) store).getHost();
             String psPath = ((PrimaryDataStoreTO) store).getPath();
-            return "nfs://" + psHost + File.separator + psPath;
+            if (StoragePoolType.NetworkFilesystem.equals(poolType)) {
+                return "nfs://" + psHost + File.separator + psPath;
+            } else if (StoragePoolType.Filesystem.equals(poolType)) {
+                return StoragePoolType.Filesystem.toString().toLowerCase() + "://" + psHost + File.separator + psPath;
+            }
         }
         return store.getUrl();
     }
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
index 2b207b4618d..20a32126992 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java
@@ -128,6 +128,7 @@ import com.cloud.network.dao.IPAddressVO;
 import com.cloud.network.dao.NetworkDao;
 import com.cloud.network.dao.NetworkVO;
 import com.cloud.network.dao.PhysicalNetworkDao;
+import com.cloud.network.router.NetworkHelper;
 import com.cloud.network.rules.FirewallRule;
 import com.cloud.network.rules.FirewallRuleVO;
 import com.cloud.network.security.SecurityGroupManager;
@@ -253,6 +254,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     private SecurityGroupManager securityGroupManager;
     @Inject
     public SecurityGroupService securityGroupService;
+    @Inject
+    public NetworkHelper networkHelper;
 
     @Inject
     private UserVmService userVmService;
@@ -360,8 +363,12 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
 
     public VMTemplateVO getKubernetesServiceTemplate(DataCenter dataCenter, Hypervisor.HypervisorType hypervisorType) {
         VMTemplateVO template = templateDao.findSystemVMReadyTemplate(dataCenter.getId(), hypervisorType);
+        if (DataCenter.Type.Edge.equals(dataCenter.getType()) && template != null && !template.isDirectDownload()) {
+            LOGGER.debug(String.format("Template %s can not be used for edge zone %s", template, dataCenter));
+            template = templateDao.findRoutingTemplate(hypervisorType, networkHelper.getHypervisorRouterTemplateConfigMap().get(hypervisorType).valueIn(dataCenter.getId()));
+        }
         if (template == null) {
-            throw new CloudRuntimeException("Not able to find the System templates or not downloaded in zone " + dataCenter.getId());
+            throw new CloudRuntimeException("Not able to find the System or Routing template in ready state for the zone " + dataCenter.getUuid());
         }
         return  template;
     }
@@ -628,7 +635,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         }
     }
 
-    private DataCenter validateAndGetZoneForKubernetesCreateParameters(Long zoneId) {
+    private DataCenter validateAndGetZoneForKubernetesCreateParameters(Long zoneId, Long networkId) {
         DataCenter zone = dataCenterDao.findById(zoneId);
         if (zone == null) {
             throw new InvalidParameterValueException("Unable to find zone by ID: " + zoneId);
@@ -636,6 +643,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         if (zone.getAllocationState() == Grouping.AllocationState.Disabled) {
             throw new PermissionDeniedException(String.format("Cannot perform this operation, zone ID: %s is currently disabled", zone.getUuid()));
         }
+        if (DataCenter.Type.Edge.equals(zone.getType()) && networkId == null) {
+            throw new PermissionDeniedException("Kubernetes clusters cannot be created on an edge zone without an existing network");
+        }
         return zone;
     }
 
@@ -675,7 +685,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
             throw new InvalidParameterValueException("Invalid name for the Kubernetes cluster name: " + name);
         }
 
-        validateAndGetZoneForKubernetesCreateParameters(zoneId);
+        validateAndGetZoneForKubernetesCreateParameters(zoneId, networkId);
         validateSshKeyPairForKubernetesCreateParameters(sshKeyPair, owner);
 
         if (nodeRootDiskSize != null && nodeRootDiskSize <= 0) {
@@ -751,7 +761,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                 String.format("Maximum cluster size can not exceed %d. Please contact your administrator", maxClusterSize));
         }
 
-        DataCenter zone = validateAndGetZoneForKubernetesCreateParameters(zoneId);
+        DataCenter zone = validateAndGetZoneForKubernetesCreateParameters(zoneId, networkId);
 
         if (!isKubernetesServiceConfigured(zone)) {
             throw new CloudRuntimeException("Kubernetes service has not been configured properly to provision Kubernetes clusters");
diff --git a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/version/KubernetesVersionManagerImpl.java b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/version/KubernetesVersionManagerImpl.java
index 643044f78d8..3ea30291f43 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/version/KubernetesVersionManagerImpl.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/version/KubernetesVersionManagerImpl.java
@@ -31,10 +31,12 @@ import org.apache.cloudstack.api.command.user.iso.RegisterIsoCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.version.ListKubernetesSupportedVersionsCmd;
 import org.apache.cloudstack.api.response.KubernetesSupportedVersionResponse;
 import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 
 import com.cloud.api.query.dao.TemplateJoinDao;
 import com.cloud.api.query.vo.TemplateJoinVO;
+import com.cloud.dc.DataCenter;
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.event.ActionEvent;
@@ -56,7 +58,6 @@ import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.exception.CloudRuntimeException;
-import org.apache.commons.lang3.StringUtils;
 
 public class KubernetesVersionManagerImpl extends ManagerBase implements KubernetesVersionService {
     public static final Logger LOGGER = Logger.getLogger(KubernetesVersionManagerImpl.class.getName());
@@ -104,6 +105,7 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
             response.setIsoId(template.getUuid());
             response.setIsoName(template.getName());
             response.setIsoState(template.getState().toString());
+            response.setDirectDownload(template.isDirectDownload());
         }
         response.setCreated(kubernetesSupportedVersion.getCreated());
         return response;
@@ -147,7 +149,7 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
         return versions;
     }
 
-    private VirtualMachineTemplate registerKubernetesVersionIso(final Long zoneId, final String versionName, final String isoUrl, final String isoChecksum)throws IllegalAccessException, NoSuchFieldException,
+    private VirtualMachineTemplate registerKubernetesVersionIso(final Long zoneId, final String versionName, final String isoUrl, final String isoChecksum, final boolean directDownload) throws IllegalAccessException, NoSuchFieldException,
             IllegalArgumentException, ResourceAllocationException {
         String isoName = String.format("%s-Kubernetes-Binaries-ISO", versionName);
         RegisterIsoCmd registerIsoCmd = new RegisterIsoCmd();
@@ -163,6 +165,7 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
         if (StringUtils.isNotEmpty(isoChecksum)) {
             registerIsoCmd.setChecksum(isoChecksum);
         }
+        registerIsoCmd.setDirectDownload(directDownload);
         registerIsoCmd.setAccountName(accountManager.getSystemAccount().getAccountName());
         registerIsoCmd.setDomainId(accountManager.getSystemAccount().getDomainId());
         return templateService.registerIso(registerIsoCmd);
@@ -288,6 +291,7 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
         final String isoChecksum = cmd.getChecksum();
         final Integer minimumCpu = cmd.getMinimumCpu();
         final Integer minimumRamSize = cmd.getMinimumRamSize();
+        final boolean isDirectDownload = cmd.isDirectDownload();
         if (minimumCpu == null || minimumCpu < KubernetesClusterService.MIN_KUBERNETES_CLUSTER_NODE_CPU) {
             throw new InvalidParameterValueException(String.format("Invalid value for %s parameter. Minimum %d vCPUs required.", ApiConstants.MIN_CPU_NUMBER, KubernetesClusterService.MIN_KUBERNETES_CLUSTER_NODE_CPU));
         }
@@ -297,8 +301,14 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
         if (compareSemanticVersions(semanticVersion, MIN_KUBERNETES_VERSION) < 0) {
             throw new InvalidParameterValueException(String.format("New supported Kubernetes version cannot be added as %s is minimum version supported by Kubernetes Service", MIN_KUBERNETES_VERSION));
         }
-        if (zoneId != null && dataCenterDao.findById(zoneId) == null) {
-            throw new InvalidParameterValueException("Invalid zone specified");
+        if (zoneId != null) {
+            DataCenter zone = dataCenterDao.findById(zoneId);
+            if (zone == null) {
+                throw new InvalidParameterValueException("Invalid zone specified");
+            }
+            if (DataCenter.Type.Edge.equals(zone.getType()) && !isDirectDownload) {
+                throw new InvalidParameterValueException(String.format("Zone: %s supports only direct download Kubernetes versions", zone.getName()));
+            }
         }
         if (StringUtils.isEmpty(isoUrl)) {
             throw new InvalidParameterValueException(String.format("Invalid URL for ISO specified, %s", isoUrl));
@@ -312,7 +322,7 @@ public class KubernetesVersionManagerImpl extends ManagerBase implements Kuberne
 
         VMTemplateVO template = null;
         try {
-            VirtualMachineTemplate vmTemplate = registerKubernetesVersionIso(zoneId, name, isoUrl, isoChecksum);
+            VirtualMachineTemplate vmTemplate = registerKubernetesVersionIso(zoneId, name, isoUrl, isoChecksum, isDirectDownload);
             template = templateDao.findById(vmTemplate.getId());
         } catch (IllegalAccessException | NoSuchFieldException | IllegalArgumentException | ResourceAllocationException ex) {
             LOGGER.error(String.format("Unable to register binaries ISO for supported kubernetes version, %s, with url: %s", name, isoUrl), ex);
diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/admin/kubernetes/version/AddKubernetesSupportedVersionCmd.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/admin/kubernetes/version/AddKubernetesSupportedVersionCmd.java
index 48bb26c8d44..380c93cca20 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/admin/kubernetes/version/AddKubernetesSupportedVersionCmd.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/admin/kubernetes/version/AddKubernetesSupportedVersionCmd.java
@@ -31,6 +31,7 @@ import org.apache.cloudstack.api.command.admin.AdminCmd;
 import org.apache.cloudstack.api.response.KubernetesSupportedVersionResponse;
 import org.apache.cloudstack.api.response.ZoneResponse;
 import org.apache.cloudstack.context.CallContext;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 
 import com.cloud.exception.ConcurrentOperationException;
@@ -38,7 +39,6 @@ import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.kubernetes.version.KubernetesSupportedVersion;
 import com.cloud.kubernetes.version.KubernetesVersionService;
 import com.cloud.utils.exception.CloudRuntimeException;
-import org.apache.commons.lang3.StringUtils;
 
 @APICommand(name = "addKubernetesSupportedVersion",
         description = "Add a supported Kubernetes version",
@@ -84,6 +84,10 @@ public class AddKubernetesSupportedVersionCmd extends BaseCmd implements AdminCm
             description = "the minimum RAM size in MB to be set with the Kubernetes version")
     private Integer minimumRamSize;
 
+    @Parameter(name=ApiConstants.DIRECT_DOWNLOAD, type = CommandType.BOOLEAN, since="4.18.2",
+            description = "If set to true the Kubernetes supported version ISO will bypass Secondary Storage and be downloaded to Primary Storage on deployment. Default is false")
+    private Boolean directDownload;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -123,6 +127,10 @@ public class AddKubernetesSupportedVersionCmd extends BaseCmd implements AdminCm
         return minimumRamSize;
     }
 
+    public boolean isDirectDownload() {
+        return (directDownload != null) ? directDownload : false;
+    }
+
     @Override
     public long getEntityOwnerId() {
         return CallContext.current().getCallingAccountId();
diff --git a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesSupportedVersionResponse.java b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesSupportedVersionResponse.java
index 72a52682364..188328ac008 100644
--- a/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesSupportedVersionResponse.java
+++ b/plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesSupportedVersionResponse.java
@@ -86,6 +86,10 @@ public class KubernetesSupportedVersionResponse extends BaseResponse {
     @Param(description = "the date when this Kubernetes supported version was created")
     private Date created;
 
+    @SerializedName(ApiConstants.DIRECT_DOWNLOAD)
+    @Param(description = "KVM Only: true if the ISO for the Kubernetes supported version is directly downloaded to Primary Storage bypassing Secondary Storage", since = "4.18.2")
+    private Boolean directDownload;
+
     public String getId() {
         return id;
     }
@@ -193,4 +197,8 @@ public class KubernetesSupportedVersionResponse extends BaseResponse {
     public void setCreated(Date created) {
         this.created = created;
     }
+
+    public void setDirectDownload(Boolean directDownload) {
+        this.directDownload = directDownload;
+    }
 }
diff --git a/server/src/main/java/com/cloud/network/router/NetworkHelper.java b/server/src/main/java/com/cloud/network/router/NetworkHelper.java
index ea008e4c4ca..c9daa5eedb4 100644
--- a/server/src/main/java/com/cloud/network/router/NetworkHelper.java
+++ b/server/src/main/java/com/cloud/network/router/NetworkHelper.java
@@ -20,6 +20,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.network.router.deployment.RouterDeploymentDefinition;
 
 import com.cloud.agent.api.to.NicTO;
@@ -93,4 +94,6 @@ public interface NetworkHelper {
             throws ConcurrentOperationException, InsufficientAddressCapacityException;
 
     public boolean validateHAProxyLBRule(final LoadBalancingRule rule);
+
+    public Map<HypervisorType, ConfigKey<String>> getHypervisorRouterTemplateConfigMap();
 }
diff --git a/server/src/main/java/com/cloud/network/router/NetworkHelperImpl.java b/server/src/main/java/com/cloud/network/router/NetworkHelperImpl.java
index 85bd43617b0..38286b5d4d9 100644
--- a/server/src/main/java/com/cloud/network/router/NetworkHelperImpl.java
+++ b/server/src/main/java/com/cloud/network/router/NetworkHelperImpl.java
@@ -28,7 +28,6 @@ import java.util.Map;
 import javax.annotation.PostConstruct;
 import javax.inject.Inject;
 
-import com.cloud.utils.validation.ChecksumUtil;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
@@ -102,6 +101,7 @@ import com.cloud.user.dao.UserDao;
 import com.cloud.utils.Pair;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.net.NetUtils;
+import com.cloud.utils.validation.ChecksumUtil;
 import com.cloud.vm.DomainRouterVO;
 import com.cloud.vm.Nic;
 import com.cloud.vm.NicProfile;
@@ -969,4 +969,9 @@ public class NetworkHelperImpl implements NetworkHelper {
     public String acquireGuestIpAddressForVrouterRedundant(Network network) {
         return _ipAddrMgr.acquireGuestIpAddressByPlacement(network, null);
     }
+
+    @Override
+    public Map<HypervisorType, ConfigKey<String>> getHypervisorRouterTemplateConfigMap() {
+        return hypervisorsMap;
+    }
 }
diff --git a/ui/src/views/compute/CreateKubernetesCluster.vue b/ui/src/views/compute/CreateKubernetesCluster.vue
index 5c725b7dd98..1f18f3cbb18 100644
--- a/ui/src/views/compute/CreateKubernetesCluster.vue
+++ b/ui/src/views/compute/CreateKubernetesCluster.vue
@@ -342,7 +342,6 @@ export default {
         const listZones = json.listzonesresponse.zone
         if (listZones) {
           this.zones = this.zones.concat(listZones)
-          this.zones = this.zones.filter(zone => zone.type !== 'Edge')
         }
       }).finally(() => {
         this.zoneLoading = false
@@ -355,6 +354,7 @@ export default {
     handleZoneChange (zone) {
       this.selectedZone = zone
       this.fetchKubernetesVersionData()
+      this.fetchNetworkData()
     },
     fetchKubernetesVersionData () {
       this.kubernetesVersions = []
@@ -413,10 +413,14 @@ export default {
     },
     fetchNetworkData () {
       const params = {}
+      if (!this.isObjectEmpty(this.selectedZone)) {
+        params.zoneid = this.selectedZone.id
+      }
       this.networkLoading = true
       api('listNetworks', params).then(json => {
-        const listNetworks = json.listnetworksresponse.network
+        var listNetworks = json.listnetworksresponse.network
         if (this.arrayHasItems(listNetworks)) {
+          listNetworks = listNetworks.filter(n => n.type !== 'L2')
           this.networks = this.networks.concat(listNetworks)
         }
       }).finally(() => {
diff --git a/ui/src/views/compute/ScaleVM.vue b/ui/src/views/compute/ScaleVM.vue
index 1a1d1309720..4ae02e7a80b 100644
--- a/ui/src/views/compute/ScaleVM.vue
+++ b/ui/src/views/compute/ScaleVM.vue
@@ -221,16 +221,19 @@ export default {
 
       this.params.serviceofferingid = id
       this.selectedOffering = this.offeringsMap[id]
-      api('listDiskOfferings', {
-        id: this.selectedOffering.diskofferingid
-      }).then(response => {
-        const diskOfferings = response.listdiskofferingsresponse.diskoffering || []
-        if (this.offerings) {
-          this.selectedDiskOffering = diskOfferings[0]
-        }
-      }).catch(error => {
-        this.$notifyError(error)
-      })
+      this.selectedDiskOffering = null
+      if (this.selectedOffering.diskofferingid) {
+        api('listDiskOfferings', {
+          id: this.selectedOffering.diskofferingid
+        }).then(response => {
+          const diskOfferings = response.listdiskofferingsresponse.diskoffering || []
+          if (this.diskOfferings) {
+            this.selectedDiskOffering = diskOfferings[0]
+          }
+        }).catch(error => {
+          this.$notifyError(error)
+        })
+      }
       this.params.automigrate = this.autoMigrate
     },
     updateFieldValue (name, value) {
diff --git a/ui/src/views/image/AddKubernetesSupportedVersion.vue b/ui/src/views/image/AddKubernetesSupportedVersion.vue
index ad4a9490a37..c88fc34695d 100644
--- a/ui/src/views/image/AddKubernetesSupportedVersion.vue
+++ b/ui/src/views/image/AddKubernetesSupportedVersion.vue
@@ -54,7 +54,8 @@
               return  option.label.toLowerCase().indexOf(input.toLowerCase()) >= 0
             }"
             :loading="zoneLoading"
-            :placeholder="apiParams.zoneid.description">
+            :placeholder="apiParams.zoneid.description"
+            @change="handleZoneChange">
             <a-select-option v-for="(opt, optIndex) in this.zones" :key="optIndex" :label="opt.name || opt.description">
               <span>
                 <resource-icon v-if="opt.icon" :image="opt.icon.base64image" size="1x" style="margin-right: 5px"/>
@@ -96,6 +97,15 @@
             v-model:value="form.minmemory"
             :placeholder="apiParams.minmemory.description"/>
         </a-form-item>
+        <a-form-item ref="directdownload" name="directdownload">
+          <template #label>
+            <tooltip-label :title="$t('label.directdownload')" :tooltip="apiParams.directdownload.description"/>
+          </template>
+          <a-switch
+            :disabled="directDownloadDisabled"
+            v-model:checked="form.directdownload"
+            :placeholder="apiParams.directdownload.description"/>
+        </a-form-item>
 
         <div :span="24" class="action-button">
           <a-button @click="closeAction">{{ $t('label.cancel') }}</a-button>
@@ -122,7 +132,10 @@ export default {
     return {
       zones: [],
       zoneLoading: false,
-      loading: false
+      loading: false,
+      selectedZone: {},
+      directDownloadDisabled: false,
+      lastNonEdgeDirectDownloadUserSelection: false
     }
   },
   beforeCreate () {
@@ -143,7 +156,8 @@ export default {
       this.formRef = ref()
       this.form = reactive({
         mincpunumber: 2,
-        minmemory: 2048
+        minmemory: 2048,
+        directdownload: false
       })
       this.rules = reactive({
         semanticversion: [{ required: true, message: this.$t('message.error.kuberversion') }],
@@ -198,7 +212,6 @@ export default {
         const listZones = json.listzonesresponse.zone
         if (listZones) {
           this.zones = this.zones.concat(listZones)
-          this.zones = this.zones.filter(zone => zone.type !== 'Edge')
         }
       }).finally(() => {
         this.zoneLoading = false
@@ -207,23 +220,38 @@ export default {
         }
       })
     },
+    handleZoneChange (zoneIdx) {
+      const zone = this.zones[zoneIdx]
+      if (this.selectedZone.type === zone.type) {
+        return
+      }
+      var lastZoneType = this.selectedZone?.type || ''
+      if (lastZoneType !== 'Edge') {
+        this.nonEdgeDirectDownloadUserSelection = this.form.directdownload
+      }
+      this.selectedZone = zone
+      if (zone.type && zone.type === 'Edge') {
+        this.form.directdownload = true
+        this.directDownloadDisabled = true
+        return
+      }
+      this.directDownloadDisabled = false
+      this.form.directdownload = this.nonEdgeDirectDownloadUserSelection
+    },
     handleSubmit (e) {
       e.preventDefault()
       if (this.loading) return
       this.formRef.value.validate().then(() => {
         const values = toRaw(this.form)
         this.loading = true
-        const params = {
-          semanticversion: values.semanticversion,
-          url: values.url
+        const params = {}
+        const customCheckParams = ['mincpunumber', 'minmemory', 'zoneid']
+        for (const key in values) {
+          if (!customCheckParams.includes(key) && values[key]) {
+            params[key] = values[key]
+          }
         }
-        if (this.isValidValueForKey(values, 'name')) {
-          params.name = values.name
-        }
-        if (this.isValidValueForKey(values, 'checksum')) {
-          params.checksum = values.checksum
-        }
-        if (this.isValidValueForKey(values, 'zoneid')) {
+        if (this.isValidValueForKey(values, 'zoneid') && values.zoneid > 0) {
           params.zoneid = this.zones[values.zoneid].id
         }
         if (this.isValidValueForKey(values, 'mincpunumber') && values.mincpunumber > 0) {