cks: Add unmanaged kubernetes cluster (#7515)

There are tools like cluster-api which create and manage kubernetes cluster on CloudStack. This PR adds the option to add unmanaged kubernetes cluster which are not managed by CKS plugin. This helps provide a consolidated view of unmanaged clusters on CloudStack. The changes done make sure that operations for managed clusters are not executed for unmanaged clusters.

Two new APIs have also been added:

1. addVirtualMachinesToKubernetesCluster - to add VMs to unmanaged clusters.
2. removeVirtualMachinesFromKubernetesCluster - to remove VMs to unmanaged clusters.

Two APIs have been updated:

1. createKubernetesCluster - made KUBERNETES_VERSION_ID, SERVICE_OFFERING_ID, SIZE as not required for unmanaged clusters. Add an additional parameter, managed, which is true by default.
2. listKubernetesClusters - Add a parameter managed to filter on managed field.

Co-authored-by: Pearl Dsilva <pearl1594@gmail.com>
Co-authored-by: dahn <daan.hoogland@gmail.com>

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -248,6 +248,7 @@ public class ApiConstants {
     public static final String IP_AVAILABLE = "ipavailable";
     public static final String IP_LIMIT = "iplimit";
     public static final String IP_TOTAL = "iptotal";
+    public static final String IS_CONTROL_NODE = "iscontrolnode";
     public static final String IS_CLEANUP_REQUIRED = "iscleanuprequired";
     public static final String IS_DYNAMIC = "isdynamic";
     public static final String IS_EDGE = "isedge";

engine/schema/src/main/resources/META-INF/db/schema-41810to41900.sql

@@ -176,3 +176,7 @@ CREATE TABLE `cloud`.`vm_scheduled_job` (
   CONSTRAINT `fk_vm_scheduled_job__vm_id` FOREIGN KEY (`vm_id`) REFERENCES `vm_instance`(`id`) ON DELETE CASCADE,
   CONSTRAINT `fk_vm_scheduled_job__vm_schedule_id` FOREIGN KEY (`vm_schedule_id`) REFERENCES `vm_schedule`(`id`) ON DELETE CASCADE
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+-- Add support for different cluster types for kubernetes
+ALTER TABLE `cloud`.`kubernetes_cluster` ADD COLUMN `cluster_type` varchar(64) DEFAULT 'CloudManaged' COMMENT 'type of cluster';
+ALTER TABLE `cloud`.`kubernetes_cluster` MODIFY COLUMN `kubernetes_version_id` bigint unsigned NULL COMMENT 'the ID of the Kubernetes version of this Kubernetes cluster';

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesCluster.java

@@ -32,6 +32,10 @@ import com.cloud.utils.fsm.StateMachine2;
  */
 public interface KubernetesCluster extends ControlledEntity, com.cloud.utils.fsm.StateObject<KubernetesCluster.State>, Identity, InternalIdentity, Displayable {
 
+    enum ClusterType {
+        CloudManaged, ExternalManaged;
+    };
+
     enum Event {
         StartRequested,
         StopRequested,
@@ -115,10 +119,10 @@ public interface KubernetesCluster extends ControlledEntity, com.cloud.utils.fsm
     String getName();
     String getDescription();
     long getZoneId();
-    long getKubernetesVersionId();
+    Long getKubernetesVersionId();
-    long getServiceOfferingId();
+    Long getServiceOfferingId();
-    long getTemplateId();
+    Long getTemplateId();
-    long getNetworkId();
+    Long getNetworkId();
     long getDomainId();
     long getAccountId();
     long getControlNodeCount();
@@ -137,4 +141,5 @@ public interface KubernetesCluster extends ControlledEntity, com.cloud.utils.fsm
     Long getMinSize();
     Long getMaxSize();
     Long getSecurityGroupId();
+    ClusterType getClusterType();
 }

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImpl.java

@@ -17,6 +17,7 @@
 package com.cloud.kubernetes.cluster;
 
 import static com.cloud.utils.NumbersUtil.toHumanReadableSize;
+import static com.cloud.vm.UserVmManager.AllowUserExpungeRecoverVm;
 
 import java.net.MalformedURLException;
 import java.net.URL;
@@ -25,8 +26,11 @@ import java.util.Arrays;
 import java.util.Date;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -36,17 +40,22 @@ import java.util.concurrent.TimeUnit;
 import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 
+import com.cloud.uservm.UserVm;
+import com.cloud.vm.UserVmService;
 import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.acl.SecurityChecker;
 import org.apache.cloudstack.annotation.AnnotationService;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.ApiConstants.VMDetails;
+import org.apache.cloudstack.api.BaseCmd;
 import org.apache.cloudstack.api.ResponseObject.ResponseView;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.AddVirtualMachinesToKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.CreateKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.DeleteKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.GetKubernetesClusterConfigCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.ListKubernetesClustersCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.RemoveVirtualMachinesFromKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.ScaleKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.StartKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.StopKubernetesClusterCmd;
@@ -54,6 +63,7 @@ import org.apache.cloudstack.api.command.user.kubernetes.cluster.UpgradeKubernet
 import org.apache.cloudstack.api.response.KubernetesClusterConfigResponse;
 import org.apache.cloudstack.api.response.KubernetesClusterResponse;
 import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.cloudstack.api.response.RemoveVirtualMachinesFromKubernetesClusterResponse;
 import org.apache.cloudstack.api.response.UserVmResponse;
 import org.apache.cloudstack.config.ApiServiceConfiguration;
 import org.apache.cloudstack.context.CallContext;
@@ -244,6 +254,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     @Inject
     public SecurityGroupService securityGroupService;
 
+    @Inject
+    private UserVmService userVmService;
+
     private void logMessage(final Level logLevel, final String message, final Exception e) {
         if (logLevel == Level.WARN) {
             if (e != null) {
@@ -535,10 +548,14 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         response.setControlNodes(kubernetesCluster.getControlNodeCount());
         response.setClusterSize(kubernetesCluster.getNodeCount());
         VMTemplateVO template = ApiDBUtils.findTemplateById(kubernetesCluster.getTemplateId());
-        response.setTemplateId(template.getUuid());
+        if (template != null) {
+            response.setTemplateId(template.getUuid());
+        }
         ServiceOfferingVO offering = serviceOfferingDao.findById(kubernetesCluster.getServiceOfferingId());
-        response.setServiceOfferingId(offering.getUuid());
+        if (offering != null) {
-        response.setServiceOfferingName(offering.getName());
+            response.setServiceOfferingId(offering.getUuid());
+            response.setServiceOfferingName(offering.getName());
+        }
         KubernetesSupportedVersionVO version = kubernetesSupportedVersionDao.findById(kubernetesCluster.getKubernetesVersionId());
         if (version != null) {
             response.setKubernetesVersionId(version.getUuid());
@@ -561,13 +578,15 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         response.setMemory(String.valueOf(kubernetesCluster.getMemory()));
         NetworkVO ntwk = networkDao.findByIdIncludingRemoved(kubernetesCluster.getNetworkId());
         response.setEndpoint(kubernetesCluster.getEndpoint());
-        response.setNetworkId(ntwk.getUuid());
+        if (ntwk != null) {
-        response.setAssociatedNetworkName(ntwk.getName());
+            response.setNetworkId(ntwk.getUuid());
-        if (ntwk.getGuestType() == Network.GuestType.Isolated) {
+            response.setAssociatedNetworkName(ntwk.getName());
-            List<IPAddressVO> ipAddresses = ipAddressDao.listByAssociatedNetwork(ntwk.getId(), true);
+            if (ntwk.getGuestType() == Network.GuestType.Isolated) {
-            if (ipAddresses != null && ipAddresses.size() == 1) {
+                List<IPAddressVO> ipAddresses = ipAddressDao.listByAssociatedNetwork(ntwk.getId(), true);
-                response.setIpAddress(ipAddresses.get(0).getAddress().addr());
+                if (ipAddresses != null && ipAddresses.size() == 1) {
-                response.setIpAddressId(ipAddresses.get(0).getUuid());
+                    response.setIpAddress(ipAddresses.get(0).getAddress().addr());
+                    response.setIpAddressId(ipAddresses.get(0).getUuid());
+                }
             }
         }
 
@@ -595,6 +614,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         response.setAutoscalingEnabled(kubernetesCluster.getAutoscalingEnabled());
         response.setMinSize(kubernetesCluster.getMinSize());
         response.setMaxSize(kubernetesCluster.getMaxSize());
+        response.setClusterType(kubernetesCluster.getClusterType());
         response.setCreated(kubernetesCluster.getCreated());
         return response;
     }
@@ -608,7 +628,95 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         }
     }
 
-    private void validateKubernetesClusterCreateParameters(final CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
+    private DataCenter validateAndGetZoneForKubernetesCreateParameters(Long zoneId) {
+        DataCenter zone = dataCenterDao.findById(zoneId);
+        if (zone == null) {
+            throw new InvalidParameterValueException("Unable to find zone by ID: " + zoneId);
+        }
+        if (zone.getAllocationState() == Grouping.AllocationState.Disabled) {
+            throw new PermissionDeniedException(String.format("Cannot perform this operation, zone ID: %s is currently disabled", zone.getUuid()));
+        }
+        return zone;
+    }
+
+    private void validateSshKeyPairForKubernetesCreateParameters(String sshKeyPair, Account owner) {
+        if (!StringUtils.isBlank(sshKeyPair)) {
+            SSHKeyPairVO sshKeyPairVO = sshKeyPairDao.findByName(owner.getAccountId(), owner.getDomainId(), sshKeyPair);
+            if (sshKeyPairVO == null) {
+                throw new InvalidParameterValueException(String.format("Given SSH key pair with name: %s was not found for the account %s", sshKeyPair, owner.getAccountName()));
+            }
+        }
+    }
+
+    private Network validateAndGetNetworkForKubernetesCreateParameters(Long networkId) {
+        Network network = null;
+        if (networkId != null) {
+            network = networkService.getNetwork(networkId);
+            if (network == null) {
+                throw new InvalidParameterValueException("Unable to find network with given ID");
+            }
+        }
+        return network;
+    }
+
+    private void validateUnmanagedKubernetesClusterCreateParameters(final CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
+        final String name = cmd.getName();
+        final Long zoneId = cmd.getZoneId();
+        final Account owner = accountService.getActiveAccountById(cmd.getEntityOwnerId());
+        final Long networkId = cmd.getNetworkId();
+        final String sshKeyPair = cmd.getSSHKeyPairName();
+        final String dockerRegistryUserName = cmd.getDockerRegistryUserName();
+        final String dockerRegistryPassword = cmd.getDockerRegistryPassword();
+        final String dockerRegistryUrl = cmd.getDockerRegistryUrl();
+        final Long nodeRootDiskSize = cmd.getNodeRootDiskSize();
+        final String externalLoadBalancerIpAddress = cmd.getExternalLoadBalancerIpAddress();
+
+        if (name == null || name.isEmpty()) {
+            throw new InvalidParameterValueException("Invalid name for the Kubernetes cluster name: " + name);
+        }
+
+        validateAndGetZoneForKubernetesCreateParameters(zoneId);
+        validateSshKeyPairForKubernetesCreateParameters(sshKeyPair, owner);
+
+        if (nodeRootDiskSize != null && nodeRootDiskSize <= 0) {
+            throw new InvalidParameterValueException(String.format("Invalid value for %s", ApiConstants.NODE_ROOT_DISK_SIZE));
+        }
+
+        validateDockerRegistryParams(dockerRegistryUserName, dockerRegistryPassword, dockerRegistryUrl);
+
+        validateAndGetNetworkForKubernetesCreateParameters(networkId);
+
+        if (StringUtils.isNotEmpty(externalLoadBalancerIpAddress) && (!NetUtils.isValidIp4(externalLoadBalancerIpAddress) && !NetUtils.isValidIp6(externalLoadBalancerIpAddress))) {
+            throw new InvalidParameterValueException("Invalid external load balancer IP address");
+        }
+    }
+
+    public boolean isCommandSupported(KubernetesCluster cluster, String cmdName) {
+        switch (cluster.getClusterType()) {
+            case CloudManaged:
+                return Arrays.asList(
+                        BaseCmd.getCommandNameByClass(CreateKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(ListKubernetesClustersCmd.class),
+                        BaseCmd.getCommandNameByClass(DeleteKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(ScaleKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(StartKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(StopKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(UpgradeKubernetesClusterCmd.class)
+                ).contains(cmdName);
+            case ExternalManaged:
+                return Arrays.asList(
+                        BaseCmd.getCommandNameByClass(CreateKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(ListKubernetesClustersCmd.class),
+                        BaseCmd.getCommandNameByClass(DeleteKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(AddVirtualMachinesToKubernetesClusterCmd.class),
+                        BaseCmd.getCommandNameByClass(RemoveVirtualMachinesFromKubernetesClusterCmd.class)
+                ).contains(cmdName);
+            default:
+                return false;
+        }
+    }
+
+    private void validateManagedKubernetesClusterCreateParameters(final CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
         validateEndpointUrl();
         final String name = cmd.getName();
         final Long zoneId = cmd.getZoneId();
@@ -619,7 +727,6 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         final String sshKeyPair = cmd.getSSHKeyPairName();
         final Long controlNodeCount = cmd.getControlNodes();
         final Long clusterSize = cmd.getClusterSize();
-        final long totalNodeCount = controlNodeCount + clusterSize;
         final String dockerRegistryUserName = cmd.getDockerRegistryUserName();
         final String dockerRegistryPassword = cmd.getDockerRegistryPassword();
         final String dockerRegistryUrl = cmd.getDockerRegistryUrl();
@@ -627,31 +734,24 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         final String externalLoadBalancerIpAddress = cmd.getExternalLoadBalancerIpAddress();
 
         if (name == null || name.isEmpty()) {
-            throw new InvalidParameterValueException("Invalid name for the Kubernetes cluster name:" + name);
+            throw new InvalidParameterValueException("Invalid name for the Kubernetes cluster name: " + name);
         }
 
         if (controlNodeCount < 1) {
             throw new InvalidParameterValueException("Invalid cluster control nodes count: " + controlNodeCount);
         }
-
+        if (clusterSize == null || clusterSize < 1) {
-        if (clusterSize < 1) {
             throw new InvalidParameterValueException("Invalid cluster size: " + clusterSize);
         }
 
         int maxClusterSize = KubernetesMaxClusterSize.valueIn(owner.getId());
+        final long totalNodeCount = controlNodeCount + clusterSize;
         if (totalNodeCount > maxClusterSize) {
             throw new InvalidParameterValueException(
                 String.format("Maximum cluster size can not exceed %d. Please contact your administrator", maxClusterSize));
         }
 
-        DataCenter zone = dataCenterDao.findById(zoneId);
+        DataCenter zone = validateAndGetZoneForKubernetesCreateParameters(zoneId);
-        if (zone == null) {
-            throw new InvalidParameterValueException("Unable to find zone by ID: " + zoneId);
-        }
-
-        if (Grouping.AllocationState.Disabled == zone.getAllocationState()) {
-            throw new PermissionDeniedException(String.format("Cannot perform this operation, zone ID: %s is currently disabled", zone.getUuid()));
-        }
 
         if (!isKubernetesServiceConfigured(zone)) {
             throw new CloudRuntimeException("Kubernetes service has not been configured properly to provision Kubernetes clusters");
@@ -694,12 +794,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
             throw new InvalidParameterValueException("No service offering with ID: " + serviceOfferingId);
         }
 
-        if (sshKeyPair != null && !sshKeyPair.isEmpty()) {
+        validateSshKeyPairForKubernetesCreateParameters(sshKeyPair, owner);
-            SSHKeyPairVO sshKeyPairVO = sshKeyPairDao.findByName(owner.getAccountId(), owner.getDomainId(), sshKeyPair);
-            if (sshKeyPairVO == null) {
-                throw new InvalidParameterValueException(String.format("Given SSH key pair with name: %s was not found for the account %s", sshKeyPair, owner.getAccountName()));
-            }
-        }
 
         if (nodeRootDiskSize != null && nodeRootDiskSize <= 0) {
             throw new InvalidParameterValueException(String.format("Invalid value for %s", ApiConstants.NODE_ROOT_DISK_SIZE));
@@ -711,13 +806,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
 
         validateDockerRegistryParams(dockerRegistryUserName, dockerRegistryPassword, dockerRegistryUrl);
 
-        Network network = null;
+        Network network = validateAndGetNetworkForKubernetesCreateParameters(networkId);
-        if (networkId != null) {
-            network = networkService.getNetwork(networkId);
-            if (network == null) {
-                throw new InvalidParameterValueException("Unable to find network with given ID");
-            }
-        }
 
         if (StringUtils.isNotEmpty(externalLoadBalancerIpAddress)) {
             if (!NetUtils.isValidIp4(externalLoadBalancerIpAddress) && !NetUtils.isValidIp6(externalLoadBalancerIpAddress)) {
@@ -788,15 +877,16 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                 List<KubernetesClusterDetailsVO> details = new ArrayList<>();
                 long kubernetesClusterId = kubernetesCluster.getId();
 
-                if (Network.GuestType.Shared.equals(network.getGuestType())) {
+                if ((network != null && Network.GuestType.Shared.equals(network.getGuestType())) || kubernetesCluster.getClusterType() == KubernetesCluster.ClusterType.ExternalManaged) {
                     addKubernetesClusterDetailIfIsNotEmpty(details, kubernetesClusterId, ApiConstants.EXTERNAL_LOAD_BALANCER_IP_ADDRESS, externalLoadBalancerIpAddress, true);
                 }
 
                 addKubernetesClusterDetailIfIsNotEmpty(details, kubernetesClusterId, ApiConstants.DOCKER_REGISTRY_USER_NAME, dockerRegistryUserName, true);
                 addKubernetesClusterDetailIfIsNotEmpty(details, kubernetesClusterId, ApiConstants.DOCKER_REGISTRY_PASSWORD, dockerRegistryPassword, false);
                 addKubernetesClusterDetailIfIsNotEmpty(details, kubernetesClusterId, ApiConstants.DOCKER_REGISTRY_URL, dockerRegistryUrl, true);
-
+                if (kubernetesCluster.getClusterType() == KubernetesCluster.ClusterType.CloudManaged) {
-                details.add(new KubernetesClusterDetailsVO(kubernetesClusterId, "networkCleanup", String.valueOf(networkCleanup), true));
+                    details.add(new KubernetesClusterDetailsVO(kubernetesClusterId, "networkCleanup", String.valueOf(networkCleanup), true));
+                }
                 kubernetesClusterDetailsDao.saveDetails(details);
             }
         });
@@ -865,6 +955,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
 
         Account caller = CallContext.current().getCallingAccount();
         accountManager.checkAccess(caller, SecurityChecker.AccessType.OperateEntry, false, kubernetesCluster);
+        if (!isCommandSupported(kubernetesCluster, cmd.getActualCommandName())) {
+            throw new InvalidParameterValueException(String.format("Scale kubernetes cluster is not supported for an externally managed cluster (%s)", kubernetesCluster.getName()));
+        }
 
         final KubernetesSupportedVersion clusterVersion = kubernetesSupportedVersionDao.findById(kubernetesCluster.getKubernetesVersionId());
         if (clusterVersion == null) {
@@ -973,6 +1066,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         if (kubernetesCluster == null || kubernetesCluster.getRemoved() != null) {
             throw new InvalidParameterValueException("Invalid Kubernetes cluster ID");
         }
+        if (!isCommandSupported(kubernetesCluster, cmd.getActualCommandName())) {
+            throw new InvalidParameterValueException(String.format("Upgrade kubernetes cluster is not supported for an externally managed cluster (%s)", kubernetesCluster.getName()));
+        }
         accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, kubernetesCluster);
         if (!KubernetesCluster.State.Running.equals(kubernetesCluster.getState())) {
             throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s is not in running state", kubernetesCluster.getName()));
@@ -1032,12 +1128,62 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     }
 
     @Override
-    public KubernetesCluster createKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
+    public KubernetesCluster createUnmanagedKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
         if (!KubernetesServiceEnabled.value()) {
             logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
         }
 
-        validateKubernetesClusterCreateParameters(cmd);
+        validateUnmanagedKubernetesClusterCreateParameters(cmd);
+
+        final DataCenter zone = dataCenterDao.findById(cmd.getZoneId());
+        final long controlNodeCount = cmd.getControlNodes();
+        final long clusterSize = Objects.requireNonNullElse(cmd.getClusterSize(), 0L);
+        final ServiceOffering serviceOffering = serviceOfferingDao.findById(cmd.getServiceOfferingId());
+        final Account owner = accountService.getActiveAccountById(cmd.getEntityOwnerId());
+        final KubernetesSupportedVersion clusterKubernetesVersion = kubernetesSupportedVersionDao.findById(cmd.getKubernetesVersionId());
+
+        final Network network = networkDao.findById(cmd.getNetworkId());
+        long cores = 0;
+        long memory = 0;
+        Long serviceOfferingId = null;
+        if (serviceOffering != null) {
+            serviceOfferingId = serviceOffering.getId();
+            cores = serviceOffering.getCpu() * (controlNodeCount + clusterSize);
+            memory = serviceOffering.getRamSize() * (controlNodeCount + clusterSize);
+        }
+
+        final Long finalServiceOfferingId = serviceOfferingId;
+        final Long defaultNetworkId = network == null ? null : network.getId();
+        final Long clusterKubernetesVersionId = clusterKubernetesVersion == null ? null : clusterKubernetesVersion.getId();
+        final long finalCores = cores;
+        final long finalMemory = memory;
+        final KubernetesClusterVO cluster = Transaction.execute(new TransactionCallback<KubernetesClusterVO>() {
+            @Override
+            public KubernetesClusterVO doInTransaction(TransactionStatus status) {
+                KubernetesClusterVO newCluster = new KubernetesClusterVO(cmd.getName(), cmd.getDisplayName(), zone.getId(), clusterKubernetesVersionId,
+                        finalServiceOfferingId, null, defaultNetworkId, owner.getDomainId(),
+                        owner.getAccountId(), controlNodeCount, clusterSize, KubernetesCluster.State.Running, cmd.getSSHKeyPairName(), finalCores, finalMemory,
+                        cmd.getNodeRootDiskSize(), "", KubernetesCluster.ClusterType.ExternalManaged);
+                kubernetesClusterDao.persist(newCluster);
+                return newCluster;
+            }
+        });
+
+        addKubernetesClusterDetails(cluster, network, cmd);
+
+        if (LOGGER.isInfoEnabled()) {
+            LOGGER.info(String.format("Kubernetes cluster with name: %s and ID: %s has been created", cluster.getName(), cluster.getUuid()));
+        }
+        return cluster;
+    }
+
+    @Override
+    public KubernetesCluster createManagedKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException {
+        if (!KubernetesServiceEnabled.value()) {
+            logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
+        }
+
+        validateManagedKubernetesClusterCreateParameters(cmd);
 
         final DataCenter zone = dataCenterDao.findById(cmd.getZoneId());
         final long controlNodeCount = cmd.getControlNodes();
@@ -1086,7 +1232,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
             public KubernetesClusterVO doInTransaction(TransactionStatus status) {
                 KubernetesClusterVO newCluster = new KubernetesClusterVO(cmd.getName(), cmd.getDisplayName(), zone.getId(), clusterKubernetesVersion.getId(),
                         serviceOffering.getId(), finalTemplate.getId(), defaultNetwork.getId(), owner.getDomainId(),
-                        owner.getAccountId(), controlNodeCount, clusterSize, KubernetesCluster.State.Created, cmd.getSSHKeyPairName(), cores, memory, cmd.getNodeRootDiskSize(), "");
+                        owner.getAccountId(), controlNodeCount, clusterSize, KubernetesCluster.State.Created, cmd.getSSHKeyPairName(), cores, memory,
+                        cmd.getNodeRootDiskSize(), "", KubernetesCluster.ClusterType.CloudManaged);
                 if (zone.isSecurityGroupEnabled()) {
                     newCluster.setSecurityGroupId(finalSecurityGroupVO.getId());
                 }
@@ -1183,7 +1330,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     }
 
     @Override
-    public boolean stopKubernetesCluster(long kubernetesClusterId) throws CloudRuntimeException {
+    public boolean stopKubernetesCluster(StopKubernetesClusterCmd cmd) throws CloudRuntimeException {
+        long kubernetesClusterId = cmd.getId();
         if (!KubernetesServiceEnabled.value()) {
             logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
         }
@@ -1191,6 +1339,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         if (kubernetesCluster == null) {
             throw new InvalidParameterValueException("Failed to find Kubernetes cluster with given ID");
         }
+        if (!isCommandSupported(kubernetesCluster, cmd.getActualCommandName())) {
+            throw new InvalidParameterValueException(String.format("Stop kubernetes cluster is not supported for an externally managed cluster (%s)", kubernetesCluster.getName()));
+        }
         if (kubernetesCluster.getRemoved() != null) {
             throw new InvalidParameterValueException(String.format("Kubernetes cluster : %s is already deleted", kubernetesCluster.getName()));
         }
@@ -1213,18 +1364,51 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
     }
 
     @Override
-    public boolean deleteKubernetesCluster(Long kubernetesClusterId) throws CloudRuntimeException {
+    public boolean deleteKubernetesCluster(DeleteKubernetesClusterCmd cmd) throws CloudRuntimeException {
         if (!KubernetesServiceEnabled.value()) {
             logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
         }
+        Long kubernetesClusterId = cmd.getId();
         KubernetesClusterVO cluster = kubernetesClusterDao.findById(kubernetesClusterId);
         if (cluster == null) {
             throw new InvalidParameterValueException("Invalid cluster id specified");
         }
         accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, cluster);
-        KubernetesClusterDestroyWorker destroyWorker = new KubernetesClusterDestroyWorker(cluster, this);
+        if (cluster.getClusterType() == KubernetesCluster.ClusterType.CloudManaged) {
-        destroyWorker = ComponentContext.inject(destroyWorker);
+            KubernetesClusterDestroyWorker destroyWorker = new KubernetesClusterDestroyWorker(cluster, this);
-        return destroyWorker.destroy();
+            destroyWorker = ComponentContext.inject(destroyWorker);
+            return destroyWorker.destroy();
+        } else {
+            boolean cleanup = cmd.getCleanup();
+            boolean expunge = cmd.getExpunge();
+            if (cleanup || expunge) {
+                CallContext ctx = CallContext.current();
+
+                if (expunge && !accountManager.isAdmin(ctx.getCallingAccount().getId()) && !AllowUserExpungeRecoverVm.valueIn(cmd.getEntityOwnerId())) {
+                    throw new PermissionDeniedException("Parameter " + ApiConstants.EXPUNGE + " can be passed by Admin only. Or when the allow.user.expunge.recover.vm key is set.");
+                }
+
+                List<KubernetesClusterVmMapVO> vmMapList = kubernetesClusterVmMapDao.listByClusterId(kubernetesClusterId);
+                for (KubernetesClusterVmMapVO vmMap : vmMapList) {
+                    try {
+                        userVmService.destroyVm(vmMap.getVmId(), expunge);
+                        if (expunge) {
+                            userVmService.expungeVm(vmMap.getVmId());
+                        }
+                    } catch (Exception exception) {
+                        logMessage(Level.WARN, String.format("Failed to destroy vm %d", vmMap.getVmId()), exception);
+                    }
+                }
+            }
+            return Transaction.execute(new TransactionCallback<Boolean>() {
+                @Override
+                public Boolean doInTransaction(TransactionStatus status) {
+                    kubernetesClusterDetailsDao.removeDetails(kubernetesClusterId);
+                    kubernetesClusterVmMapDao.removeByClusterId(kubernetesClusterId);
+                    return kubernetesClusterDao.remove(kubernetesClusterId);
+                }
+            });
+        }
     }
 
     @Override
@@ -1238,6 +1422,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         final String state = cmd.getState();
         final String name = cmd.getName();
         final String keyword = cmd.getKeyword();
+        final String cmdClusterType = cmd.getClusterType();
         List<KubernetesClusterResponse> responsesList = new ArrayList<KubernetesClusterResponse>();
         List<Long> permittedAccounts = new ArrayList<Long>();
         Ternary<Long, Boolean, Project.ListProjectResourcesCriteria> domainIdRecursiveListProject = new Ternary<Long, Boolean, Project.ListProjectResourcesCriteria>(cmd.getDomainId(), cmd.isRecursive(), null);
@@ -1245,6 +1430,17 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         Long domainId = domainIdRecursiveListProject.first();
         Boolean isRecursive = domainIdRecursiveListProject.second();
         Project.ListProjectResourcesCriteria listProjectResourcesCriteria = domainIdRecursiveListProject.third();
+
+        KubernetesCluster.ClusterType clusterType = null;
+
+        if (cmdClusterType != null) {
+            try {
+             clusterType = KubernetesCluster.ClusterType.valueOf(cmdClusterType);
+            } catch (IllegalArgumentException exception) {
+                throw new InvalidParameterValueException("Unable to resolve cluster type " + cmdClusterType + " to a supported value (CloudManaged, ExternalManaged)");
+            }
+        }
+
         Filter searchFilter = new Filter(KubernetesClusterVO.class, "id", true, cmd.getStartIndex(), cmd.getPageSizeVal());
         SearchBuilder<KubernetesClusterVO> sb = kubernetesClusterDao.createSearchBuilder();
         accountManager.buildACLSearchBuilder(sb, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
@@ -1252,6 +1448,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         sb.and("name", sb.entity().getName(), SearchCriteria.Op.EQ);
         sb.and("keyword", sb.entity().getName(), SearchCriteria.Op.LIKE);
         sb.and("state", sb.entity().getState(), SearchCriteria.Op.IN);
+        sb.and("cluster_type", sb.entity().getClusterType(), SearchCriteria.Op.EQ);
         SearchCriteria<KubernetesClusterVO> sc = sb.create();
         accountManager.buildACLSearchCriteria(sc, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
         if (state != null) {
@@ -1266,6 +1463,9 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         if (name != null) {
             sc.setParameters("name", name);
         }
+        if (clusterType != null) {
+            sc.setParameters("cluster_type", clusterType);
+        }
         List<KubernetesClusterVO> kubernetesClusters = kubernetesClusterDao.search(sc, searchFilter);
         for (KubernetesClusterVO cluster : kubernetesClusters) {
             KubernetesClusterResponse clusterResponse = createKubernetesClusterResponse(cluster.getId());
@@ -1342,6 +1542,114 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         return upgradeWorker.upgradeCluster();
     }
 
+    private void updateNodeCount(KubernetesClusterVO kubernetesCluster) {
+        List<KubernetesClusterVmMapVO> nodeList = kubernetesClusterVmMapDao.listByClusterId(kubernetesCluster.getId());
+        kubernetesCluster.setControlNodeCount(nodeList.stream().filter(KubernetesClusterVmMapVO::isControlNode).count());
+        kubernetesCluster.setNodeCount(nodeList.size());
+        kubernetesCluster.setNodeCount(nodeList.size());
+        kubernetesClusterDao.persist(kubernetesCluster);
+    }
+
+    @Override
+    public boolean addVmsToCluster(AddVirtualMachinesToKubernetesClusterCmd cmd) {
+        if (!KubernetesServiceEnabled.value()) {
+            logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
+        }
+        List<Long> vmIds = cmd.getVmIds();
+        Long clusterId = cmd.getId();
+
+        KubernetesClusterVO kubernetesCluster = kubernetesClusterDao.findById(clusterId);
+        if (kubernetesCluster == null) {
+            throw new InvalidParameterValueException("Invalid Kubernetes cluster ID specified");
+        }
+        if (!isCommandSupported(kubernetesCluster, cmd.getActualCommandName())) {
+            throw new InvalidParameterValueException("VM cannot be added to a CloudStack managed Kubernetes cluster");
+        }
+
+        // User should have access to both VM and Kubernetes cluster
+        accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, kubernetesCluster);
+
+        for (Long vmId : vmIds) {
+            VMInstanceVO vmInstance = vmInstanceDao.findById(vmId);
+            if (vmInstance == null) {
+                throw new InvalidParameterValueException("Invalid VM ID specified");
+            }
+            accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, vmInstance);
+        }
+
+        KubernetesClusterVmMapVO clusterVmMap = null;
+        List<KubernetesClusterVmMapVO> clusterVmMapList = kubernetesClusterVmMapDao.listByClusterIdAndVmIdsIn(clusterId, vmIds);
+        ArrayList<Long> alreadyExistingVmIds = new ArrayList<>();
+        for (KubernetesClusterVmMapVO clusterVmMapVO : clusterVmMapList) {
+            alreadyExistingVmIds.add(clusterVmMapVO.getVmId());
+        }
+        vmIds.removeAll(alreadyExistingVmIds);
+        for (Long vmId : vmIds) {
+            clusterVmMap = new KubernetesClusterVmMapVO(clusterId, vmId, cmd.isControlNode());
+            kubernetesClusterVmMapDao.persist(clusterVmMap);
+        }
+        updateNodeCount(kubernetesCluster);
+        return true;
+    }
+
+    @Override
+    public List<RemoveVirtualMachinesFromKubernetesClusterResponse> removeVmsFromCluster(RemoveVirtualMachinesFromKubernetesClusterCmd cmd) {
+        if (!KubernetesServiceEnabled.value()) {
+            logAndThrow(Level.ERROR, "Kubernetes Service plugin is disabled");
+        }
+        List<Long> vmIds = cmd.getVmIds();
+        Long clusterId = cmd.getId();
+
+        KubernetesClusterVO kubernetesCluster = kubernetesClusterDao.findById(clusterId);
+        if (kubernetesCluster == null) {
+            throw new InvalidParameterValueException("Invalid Kubernetes cluster ID specified");
+        }
+        if (!isCommandSupported(kubernetesCluster, cmd.getActualCommandName())) {
+            throw new InvalidParameterValueException("VM cannot be removed from a CloudStack Managed Kubernetes cluster");
+        }
+        accountManager.checkAccess(CallContext.current().getCallingAccount(), SecurityChecker.AccessType.OperateEntry, false, kubernetesCluster);
+
+        List<KubernetesClusterVmMapVO> kubernetesClusterVmMap = kubernetesClusterVmMapDao.listByClusterIdAndVmIdsIn(clusterId, vmIds);
+        List<RemoveVirtualMachinesFromKubernetesClusterResponse> responseList = new ArrayList<>();
+
+        Set<Long> vmIdsRemoved = new HashSet<>();
+
+        for (KubernetesClusterVmMapVO clusterVmMap : kubernetesClusterVmMap) {
+            RemoveVirtualMachinesFromKubernetesClusterResponse response = new RemoveVirtualMachinesFromKubernetesClusterResponse();
+            UserVm vm = userVmService.getUserVm(clusterVmMap.getVmId());
+            response.setVmId(vm.getUuid());
+            response.setSuccess(kubernetesClusterVmMapDao.remove(clusterVmMap.getId()));
+            response.setObjectName(cmd.getCommandName());
+            responseList.add(response);
+            vmIdsRemoved.add(clusterVmMap.getVmId());
+        }
+
+        for (Long vmId : vmIds) {
+            if (!vmIdsRemoved.contains(vmId)) {
+                RemoveVirtualMachinesFromKubernetesClusterResponse response = new RemoveVirtualMachinesFromKubernetesClusterResponse();
+                VMInstanceVO vm = vmInstanceDao.findByIdIncludingRemoved(vmId);
+                if (vm == null) {
+                    response.setVmId(vmId.toString());
+                    response.setDisplayText("Not a valid vm id");
+                    vmIdsRemoved.add(vmId);
+                } else {
+                    response.setVmId(vm.getUuid());
+                    vmIdsRemoved.add(vmId);
+                    if (vm.isRemoved()) {
+                        response.setDisplayText("VM is already removed");
+                    } else {
+                        response.setDisplayText("VM is not part of the cluster");
+                    }
+                }
+                response.setObjectName(cmd.getCommandName());
+                response.setSuccess(false);
+                responseList.add(response);
+            }
+        }
+        updateNodeCount(kubernetesCluster);
+        return responseList;
+    }
+
     @Override
     public List<Class<?>> getCommands() {
         List<Class<?>> cmdList = new ArrayList<Class<?>>();
@@ -1356,6 +1664,8 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         cmdList.add(GetKubernetesClusterConfigCmd.class);
         cmdList.add(ScaleKubernetesClusterCmd.class);
         cmdList.add(UpgradeKubernetesClusterCmd.class);
+        cmdList.add(AddVirtualMachinesToKubernetesClusterCmd.class);
+        cmdList.add(RemoveVirtualMachinesFromKubernetesClusterCmd.class);
         return cmdList;
     }
 
@@ -1442,7 +1752,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
         public void reallyRun() {
             try {
                 // run through Kubernetes clusters in 'Running' state and ensure all the VM's are Running in the cluster
-                List<KubernetesClusterVO> runningKubernetesClusters = kubernetesClusterDao.findKubernetesClustersInState(KubernetesCluster.State.Running);
+                List<KubernetesClusterVO> runningKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Running);
                 for (KubernetesCluster kubernetesCluster : runningKubernetesClusters) {
                     if (LOGGER.isInfoEnabled()) {
                         LOGGER.info(String.format("Running Kubernetes cluster state scanner on Kubernetes cluster : %s", kubernetesCluster.getName()));
@@ -1457,7 +1767,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                 }
 
                 // run through Kubernetes clusters in 'Stopped' state and ensure all the VM's are Stopped in the cluster
-                List<KubernetesClusterVO> stoppedKubernetesClusters = kubernetesClusterDao.findKubernetesClustersInState(KubernetesCluster.State.Stopped);
+                List<KubernetesClusterVO> stoppedKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Stopped);
                 for (KubernetesCluster kubernetesCluster : stoppedKubernetesClusters) {
                     if (LOGGER.isInfoEnabled()) {
                         LOGGER.info(String.format("Running Kubernetes cluster state scanner on Kubernetes cluster : %s for state: %s", kubernetesCluster.getName(), KubernetesCluster.State.Stopped.toString()));
@@ -1472,7 +1782,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                 }
 
                 // run through Kubernetes clusters in 'Alert' state and reconcile state as 'Running' if the VM's are running or 'Stopped' if VM's are stopped
-                List<KubernetesClusterVO> alertKubernetesClusters = kubernetesClusterDao.findKubernetesClustersInState(KubernetesCluster.State.Alert);
+                List<KubernetesClusterVO> alertKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Alert);
                 for (KubernetesClusterVO kubernetesCluster : alertKubernetesClusters) {
                     if (LOGGER.isInfoEnabled()) {
                         LOGGER.info(String.format("Running Kubernetes cluster state scanner on Kubernetes cluster : %s for state: %s", kubernetesCluster.getName(), KubernetesCluster.State.Alert.toString()));
@@ -1495,7 +1805,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
 
                 if (firstRun) {
                     // run through Kubernetes clusters in 'Starting' state and reconcile state as 'Alert' or 'Error' if the VM's are running
-                    List<KubernetesClusterVO> startingKubernetesClusters = kubernetesClusterDao.findKubernetesClustersInState(KubernetesCluster.State.Starting);
+                    List<KubernetesClusterVO> startingKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Starting);
                     for (KubernetesCluster kubernetesCluster : startingKubernetesClusters) {
                         if ((new Date()).getTime() - kubernetesCluster.getCreated().getTime() < 10*60*1000) {
                             continue;
@@ -1513,7 +1823,7 @@ public class KubernetesClusterManagerImpl extends ManagerBase implements Kuberne
                             LOGGER.warn(String.format("Failed to run Kubernetes cluster Starting state scanner on Kubernetes cluster : %s status scanner", kubernetesCluster.getName()), e);
                         }
                     }
-                    List<KubernetesClusterVO> destroyingKubernetesClusters = kubernetesClusterDao.findKubernetesClustersInState(KubernetesCluster.State.Destroying);
+                    List<KubernetesClusterVO> destroyingKubernetesClusters = kubernetesClusterDao.findManagedKubernetesClustersInState(KubernetesCluster.State.Destroying);
                     for (KubernetesCluster kubernetesCluster : destroyingKubernetesClusters) {
                         if (LOGGER.isInfoEnabled()) {
                             LOGGER.info(String.format("Running Kubernetes cluster state scanner on Kubernetes cluster : %s for state: %s", kubernetesCluster.getName(), KubernetesCluster.State.Destroying.toString()));

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterService.java

@@ -16,20 +16,27 @@
 // under the License.
 package com.cloud.kubernetes.cluster;
 
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.AddVirtualMachinesToKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.CreateKubernetesClusterCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.DeleteKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.GetKubernetesClusterConfigCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.ListKubernetesClustersCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.RemoveVirtualMachinesFromKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.ScaleKubernetesClusterCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.StopKubernetesClusterCmd;
 import org.apache.cloudstack.api.command.user.kubernetes.cluster.UpgradeKubernetesClusterCmd;
 import org.apache.cloudstack.api.response.KubernetesClusterConfigResponse;
 import org.apache.cloudstack.api.response.KubernetesClusterResponse;
 import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.cloudstack.api.response.RemoveVirtualMachinesFromKubernetesClusterResponse;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.Configurable;
 
 import com.cloud.utils.component.PluggableService;
 import com.cloud.utils.exception.CloudRuntimeException;
 
+import java.util.List;
+
 public interface KubernetesClusterService extends PluggableService, Configurable {
     static final String MIN_KUBERNETES_VERSION_HA_SUPPORT = "1.16.0";
     static final int MIN_KUBERNETES_CLUSTER_NODE_CPU = 2;
@@ -81,13 +88,17 @@ public interface KubernetesClusterService extends PluggableService, Configurable
 
     KubernetesCluster findById(final Long id);
 
-    KubernetesCluster createKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException;
+    KubernetesCluster createUnmanagedKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException;
+
+    KubernetesCluster createManagedKubernetesCluster(CreateKubernetesClusterCmd cmd) throws CloudRuntimeException;
 
     boolean startKubernetesCluster(long kubernetesClusterId, boolean onCreate) throws CloudRuntimeException;
 
-    boolean stopKubernetesCluster(long kubernetesClusterId) throws CloudRuntimeException;
+    boolean stopKubernetesCluster(StopKubernetesClusterCmd cmd) throws CloudRuntimeException;
 
-    boolean deleteKubernetesCluster(Long kubernetesClusterId) throws CloudRuntimeException;
+    boolean deleteKubernetesCluster(DeleteKubernetesClusterCmd cmd) throws CloudRuntimeException;
+
+    boolean isCommandSupported(KubernetesCluster cluster, String cmdName);
 
     ListResponse<KubernetesClusterResponse> listKubernetesClusters(ListKubernetesClustersCmd cmd);
 
@@ -98,4 +109,8 @@ public interface KubernetesClusterService extends PluggableService, Configurable
     boolean scaleKubernetesCluster(ScaleKubernetesClusterCmd cmd) throws CloudRuntimeException;
 
     boolean upgradeKubernetesCluster(UpgradeKubernetesClusterCmd cmd) throws CloudRuntimeException;
+
+    boolean addVmsToCluster(AddVirtualMachinesToKubernetesClusterCmd cmd);
+
+    List<RemoveVirtualMachinesFromKubernetesClusterResponse> removeVmsFromCluster(RemoveVirtualMachinesFromKubernetesClusterCmd cmd);
 }

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/KubernetesClusterVO.java

@@ -52,16 +52,16 @@ public class KubernetesClusterVO implements KubernetesCluster {
     private long zoneId;
 
     @Column(name = "kubernetes_version_id")
-    private long kubernetesVersionId;
+    private Long kubernetesVersionId;
 
     @Column(name = "service_offering_id")
-    private long serviceOfferingId;
+    private Long serviceOfferingId;
 
     @Column(name = "template_id")
-    private long templateId;
+    private Long templateId;
 
     @Column(name = "network_id")
-    private long networkId;
+    private Long networkId;
 
     @Column(name = "domain_id")
     private long domainId;
@@ -114,6 +114,9 @@ public class KubernetesClusterVO implements KubernetesCluster {
     @Column(name = "security_group_id")
     private Long securityGroupId;
 
+    @Column(name = "cluster_type")
+    private ClusterType clusterType;
+
     @Override
     public long getId() {
         return id;
@@ -160,7 +163,7 @@ public class KubernetesClusterVO implements KubernetesCluster {
     }
 
     @Override
-    public long getKubernetesVersionId() {
+    public Long getKubernetesVersionId() {
         return kubernetesVersionId;
     }
 
@@ -169,7 +172,7 @@ public class KubernetesClusterVO implements KubernetesCluster {
     }
 
     @Override
-    public long getServiceOfferingId() {
+    public Long getServiceOfferingId() {
         return serviceOfferingId;
     }
 
@@ -178,7 +181,7 @@ public class KubernetesClusterVO implements KubernetesCluster {
     }
 
     @Override
-    public long getTemplateId() {
+    public Long getTemplateId() {
         return templateId;
     }
 
@@ -187,7 +190,7 @@ public class KubernetesClusterVO implements KubernetesCluster {
     }
 
     @Override
-    public long getNetworkId() {
+    public Long getNetworkId() {
         return networkId;
     }
 
@@ -350,13 +353,21 @@ public class KubernetesClusterVO implements KubernetesCluster {
         return securityGroupId;
     }
 
+    public ClusterType getClusterType() {
+        return clusterType;
+    }
+
+    public void setClusterType(ClusterType clusterType) {
+        this.clusterType = clusterType;
+    }
+
     public KubernetesClusterVO() {
         this.uuid = UUID.randomUUID().toString();
     }
 
-    public KubernetesClusterVO(String name, String description, long zoneId, long kubernetesVersionId, long serviceOfferingId, long templateId,
+    public KubernetesClusterVO(String name, String description, long zoneId, Long kubernetesVersionId, Long serviceOfferingId, Long templateId,
-                               long networkId, long domainId, long accountId, long controlNodeCount, long nodeCount, State state,
+                               Long networkId, long domainId, long accountId, long controlNodeCount, long nodeCount, State state,
-                               String keyPair, long cores, long memory, Long nodeRootDiskSize, String endpoint) {
+                               String keyPair, long cores, long memory, Long nodeRootDiskSize, String endpoint, ClusterType clusterType) {
         this.uuid = UUID.randomUUID().toString();
         this.name = name;
         this.description = description;
@@ -377,14 +388,15 @@ public class KubernetesClusterVO implements KubernetesCluster {
             this.nodeRootDiskSize = nodeRootDiskSize;
         }
         this.endpoint = endpoint;
+        this.clusterType = clusterType;
         this.checkForGc = false;
     }
 
     public KubernetesClusterVO(String name, String description, long zoneId, long kubernetesVersionId, long serviceOfferingId, long templateId,
         long networkId, long domainId, long accountId, long controlNodeCount, long nodeCount, State state, String keyPair, long cores,
-        long memory, Long nodeRootDiskSize, String endpoint, boolean autoscalingEnabled, Long minSize, Long maxSize) {
+        long memory, Long nodeRootDiskSize, String endpoint, ClusterType clusterType, boolean autoscalingEnabled, Long minSize, Long maxSize) {
         this(name, description, zoneId, kubernetesVersionId, serviceOfferingId, templateId, networkId, domainId, accountId, controlNodeCount,
-            nodeCount, state, keyPair, cores, memory, nodeRootDiskSize, endpoint);
+            nodeCount, state, keyPair, cores, memory, nodeRootDiskSize, endpoint, clusterType);
         this.autoscalingEnabled = autoscalingEnabled;
         this.minSize = minSize;
         this.maxSize = maxSize;

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDao.java

@@ -28,7 +28,7 @@ public interface KubernetesClusterDao extends GenericDao<KubernetesClusterVO, Lo
 
     List<KubernetesClusterVO> listByAccount(long accountId);
     List<KubernetesClusterVO> findKubernetesClustersToGarbageCollect();
-    List<KubernetesClusterVO> findKubernetesClustersInState(KubernetesCluster.State state);
+    List<KubernetesClusterVO> findManagedKubernetesClustersInState(KubernetesCluster.State state);
     List<KubernetesClusterVO> listByNetworkId(long networkId);
     List<KubernetesClusterVO> listAllByKubernetesVersion(long kubernetesVersionId);
 }

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterDaoImpl.java

@@ -32,7 +32,7 @@ public class KubernetesClusterDaoImpl extends GenericDaoBase<KubernetesClusterVO
 
     private final SearchBuilder<KubernetesClusterVO> AccountIdSearch;
     private final SearchBuilder<KubernetesClusterVO> GarbageCollectedSearch;
-    private final SearchBuilder<KubernetesClusterVO> StateSearch;
+    private final SearchBuilder<KubernetesClusterVO> ManagedStateSearch;
     private final SearchBuilder<KubernetesClusterVO> SameNetworkSearch;
     private final SearchBuilder<KubernetesClusterVO> KubernetesVersionSearch;
 
@@ -44,11 +44,13 @@ public class KubernetesClusterDaoImpl extends GenericDaoBase<KubernetesClusterVO
         GarbageCollectedSearch = createSearchBuilder();
         GarbageCollectedSearch.and("gc", GarbageCollectedSearch.entity().isCheckForGc(), SearchCriteria.Op.EQ);
         GarbageCollectedSearch.and("state", GarbageCollectedSearch.entity().getState(), SearchCriteria.Op.EQ);
+        GarbageCollectedSearch.and("cluster_type", GarbageCollectedSearch.entity().getClusterType(), SearchCriteria.Op.EQ);
         GarbageCollectedSearch.done();
 
-        StateSearch = createSearchBuilder();
+        ManagedStateSearch = createSearchBuilder();
-        StateSearch.and("state", StateSearch.entity().getState(), SearchCriteria.Op.EQ);
+        ManagedStateSearch.and("state", ManagedStateSearch.entity().getState(), SearchCriteria.Op.EQ);
-        StateSearch.done();
+        ManagedStateSearch.and("cluster_type", ManagedStateSearch.entity().getClusterType(), SearchCriteria.Op.EQ);
+        ManagedStateSearch.done();
 
         SameNetworkSearch = createSearchBuilder();
         SameNetworkSearch.and("network_id", SameNetworkSearch.entity().getNetworkId(), SearchCriteria.Op.EQ);
@@ -71,13 +73,15 @@ public class KubernetesClusterDaoImpl extends GenericDaoBase<KubernetesClusterVO
         SearchCriteria<KubernetesClusterVO> sc = GarbageCollectedSearch.create();
         sc.setParameters("gc", true);
         sc.setParameters("state", KubernetesCluster.State.Destroying);
+        sc.setParameters("cluster_type", KubernetesCluster.ClusterType.CloudManaged);
         return listBy(sc);
     }
 
     @Override
-    public List<KubernetesClusterVO> findKubernetesClustersInState(KubernetesCluster.State state) {
+    public List<KubernetesClusterVO> findManagedKubernetesClustersInState(KubernetesCluster.State state) {
-        SearchCriteria<KubernetesClusterVO> sc = StateSearch.create();
+        SearchCriteria<KubernetesClusterVO> sc = ManagedStateSearch.create();
         sc.setParameters("state", state);
+        sc.setParameters("cluster_type", KubernetesCluster.ClusterType.CloudManaged);
         return listBy(sc);
     }
 

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterVmMapDao.java

@@ -24,4 +24,8 @@ import java.util.List;
 public interface KubernetesClusterVmMapDao extends GenericDao<KubernetesClusterVmMapVO, Long> {
     public List<KubernetesClusterVmMapVO> listByClusterId(long clusterId);
     public List<KubernetesClusterVmMapVO> listByClusterIdAndVmIdsIn(long clusterId, List<Long> vmIds);
+
+    int removeByClusterIdAndVmIdsIn(long clusterId, List<Long> vmIds);
+
+    public int removeByClusterId(long clusterId);
 }

plugins/integrations/kubernetes-service/src/main/java/com/cloud/kubernetes/cluster/dao/KubernetesClusterVmMapDaoImpl.java

@@ -54,4 +54,19 @@ public class KubernetesClusterVmMapDaoImpl extends GenericDaoBase<KubernetesClus
         sc.setParameters("vmIdsIN", vmIds.toArray());
         return listBy(sc);
     }
+
+    @Override
+    public int removeByClusterIdAndVmIdsIn(long clusterId, List<Long> vmIds) {
+        SearchCriteria<KubernetesClusterVmMapVO> sc = clusterIdSearch.create();
+        sc.setParameters("clusterId", clusterId);
+        sc.setParameters("vmIdsIN", vmIds.toArray());
+        return remove(sc);
+    }
+
+    @Override
+    public int removeByClusterId(long clusterId) {
+        SearchCriteria<KubernetesClusterVmMapVO> sc = clusterIdSearch.create();
+        sc.setParameters("clusterId", clusterId);
+        return remove(sc);
+    }
 }

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/AddVirtualMachinesToKubernetesClusterCmd.java

@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.api.command.user.kubernetes.cluster;
+
+import com.cloud.kubernetes.cluster.KubernetesClusterService;
+import com.cloud.utils.exception.CloudRuntimeException;
+import org.apache.cloudstack.acl.RoleType;
+import org.apache.cloudstack.api.APICommand;
+import org.apache.cloudstack.api.ApiConstants;
+import org.apache.cloudstack.api.ApiErrorCode;
+import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.api.Parameter;
+import org.apache.cloudstack.api.ServerApiException;
+import org.apache.cloudstack.api.response.KubernetesClusterResponse;
+import org.apache.cloudstack.api.response.SuccessResponse;
+import org.apache.cloudstack.api.response.UserVmResponse;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.log4j.Logger;
+
+import javax.inject.Inject;
+import java.util.List;
+
+@APICommand(name = "addVirtualMachinesToKubernetesCluster",
+        description = "Add VMs to an ExternalManaged kubernetes cluster. Not applicable for CloudManaged kubernetes clusters.",
+        responseObject = SuccessResponse.class,
+        since = "4.19.0",
+        authorized = {RoleType.Admin, RoleType.ResourceAdmin, RoleType.DomainAdmin, RoleType.User})
+public class AddVirtualMachinesToKubernetesClusterCmd extends BaseCmd {
+    public static final Logger LOGGER = Logger.getLogger(AddVirtualMachinesToKubernetesClusterCmd.class.getName());
+
+    @Inject
+    public KubernetesClusterService kubernetesClusterService;
+
+    /////////////////////////////////////////////////////
+    //////////////// API parameters /////////////////////
+    /////////////////////////////////////////////////////
+
+    @Parameter(name = ApiConstants.ID, type = CommandType.UUID,
+            entityType = KubernetesClusterResponse.class,
+            required = true,
+            description = "the ID of the Kubernetes cluster")
+    private Long id;
+
+    @Parameter(name = ApiConstants.VIRTUAL_MACHINE_IDS, type = CommandType.LIST,
+            collectionType=CommandType.UUID,
+            entityType = UserVmResponse.class,
+            required = true,
+            description = "the IDs of the VMs to add to the cluster")
+    private List<Long> vmIds;
+
+    @Parameter(name = ApiConstants.IS_CONTROL_NODE, type = CommandType.BOOLEAN,
+            description = "Is control node or not? Defaults to false.")
+    private Boolean isControlNode;
+
+    /////////////////////////////////////////////////////
+    /////////////////// Accessors ///////////////////////
+    /////////////////////////////////////////////////////
+
+    public Long getId() {
+        return id;
+    }
+
+    public List<Long> getVmIds() {
+        return vmIds;
+    }
+
+    public boolean isControlNode() {
+        return (isControlNode != null) && isControlNode;
+    }
+    /////////////////////////////////////////////////////
+    /////////////// API Implementation///////////////////
+    /////////////////////////////////////////////////////
+
+    @Override
+    public long getEntityOwnerId() {
+        return CallContext.current().getCallingAccount().getId();
+    }
+
+    @Override
+    public void execute() throws ServerApiException {
+        try {
+            if (!kubernetesClusterService.addVmsToCluster(this)) {
+                throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Failed to add VMs to cluster");
+            }
+            final SuccessResponse response = new SuccessResponse();
+            response.setResponseName(getCommandName());
+            setResponseObject(response);
+        } catch (CloudRuntimeException e) {
+            throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, e.getMessage());
+        }
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/CreateKubernetesClusterCmd.java

@@ -20,6 +20,7 @@ import java.security.InvalidParameterException;
 
 import javax.inject.Inject;
 
+import com.cloud.exception.InvalidParameterValueException;
 import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.acl.SecurityChecker.AccessType;
 import org.apache.cloudstack.api.ACL;
@@ -77,13 +78,13 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
             description = "availability zone in which Kubernetes cluster to be launched")
     private Long zoneId;
 
-    @Parameter(name = ApiConstants.KUBERNETES_VERSION_ID, type = CommandType.UUID, entityType = KubernetesSupportedVersionResponse.class, required = true,
+    @Parameter(name = ApiConstants.KUBERNETES_VERSION_ID, type = CommandType.UUID, entityType = KubernetesSupportedVersionResponse.class,
             description = "Kubernetes version with which cluster to be launched")
     private Long kubernetesVersionId;
 
     @ACL(accessType = AccessType.UseEntry)
     @Parameter(name = ApiConstants.SERVICE_OFFERING_ID, type = CommandType.UUID, entityType = ServiceOfferingResponse.class,
-            required = true, description = "the ID of the service offering for the virtual machines in the cluster.")
+            description = "the ID of the service offering for the virtual machines in the cluster.")
     private Long serviceOfferingId;
 
     @ACL(accessType = AccessType.UseEntry)
@@ -125,7 +126,7 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
     private String externalLoadBalancerIpAddress;
 
     @Parameter(name=ApiConstants.SIZE, type = CommandType.LONG,
-            required = true, description = "number of Kubernetes cluster worker nodes")
+            description = "number of Kubernetes cluster worker nodes")
     private Long clusterSize;
 
     @Parameter(name = ApiConstants.DOCKER_REGISTRY_USER_NAME, type = CommandType.STRING,
@@ -144,6 +145,9 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
             description = "root disk size in GB for each node")
     private Long nodeRootDiskSize;
 
+    @Parameter(name = ApiConstants.CLUSTER_TYPE, type = CommandType.STRING, required = true, description = "type of the cluster: CloudManaged, ExternalManaged", since="4.19.0")
+    private String clusterType;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -233,6 +237,13 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
         }
     }
 
+    public String getClusterType() {
+        if (clusterType == null) {
+            return KubernetesCluster.ClusterType.CloudManaged.toString();
+        }
+        return clusterType;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -279,7 +290,8 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
     @Override
     public void execute() {
         try {
-            if (!kubernetesClusterService.startKubernetesCluster(getEntityId(), true)) {
+            if (KubernetesCluster.ClusterType.valueOf(getClusterType()) == KubernetesCluster.ClusterType.CloudManaged
+                    && !kubernetesClusterService.startKubernetesCluster(getEntityId(), true)) {
                 throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Failed to start Kubernetes cluster");
             }
             KubernetesClusterResponse response = kubernetesClusterService.createKubernetesClusterResponse(getEntityId());
@@ -292,8 +304,20 @@ public class CreateKubernetesClusterCmd extends BaseAsyncCreateCmd {
 
     @Override
     public void create() throws CloudRuntimeException {
+        KubernetesCluster cluster;
+        KubernetesCluster.ClusterType type;
         try {
-            KubernetesCluster cluster = kubernetesClusterService.createKubernetesCluster(this);
+            type = KubernetesCluster.ClusterType.valueOf(getClusterType());
+        } catch (IllegalArgumentException e) {
+            throw new InvalidParameterValueException("Unable to resolve cluster type " + getClusterType() + " to a supported value (CloudManaged, ExternalManaged)");
+        }
+
+        try {
+            if (type == KubernetesCluster.ClusterType.CloudManaged) {
+                cluster = kubernetesClusterService.createManagedKubernetesCluster(this);
+            } else {
+                cluster = kubernetesClusterService.createUnmanagedKubernetesCluster(this);
+            }
             if (cluster == null) {
                 throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Failed to create Kubernetes cluster");
             }

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/DeleteKubernetesClusterCmd.java

@@ -58,6 +58,18 @@ public class DeleteKubernetesClusterCmd extends BaseAsyncCmd {
             description = "the ID of the Kubernetes cluster")
     private Long id;
 
+    @Parameter(name = ApiConstants.CLEANUP,
+            type = CommandType.BOOLEAN,
+            since = "4.19.0",
+            description = "Destroy attached instances of the ExternalManaged Cluster. Default: false")
+    private Boolean cleanup;
+
+    @Parameter(name = ApiConstants.EXPUNGE,
+            type = CommandType.BOOLEAN,
+            since = "4.19.0",
+            description = "Expunge attached instances of the ExternalManaged Cluster. If true, value of cleanup is ignored. Default: false")
+    private Boolean expunge;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -66,6 +78,14 @@ public class DeleteKubernetesClusterCmd extends BaseAsyncCmd {
         return id;
     }
 
+    public Boolean getCleanup() {
+        return cleanup != null && cleanup;
+    }
+
+    public Boolean getExpunge() {
+        return expunge != null && expunge;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -73,7 +93,7 @@ public class DeleteKubernetesClusterCmd extends BaseAsyncCmd {
     @Override
     public void execute() throws ServerApiException, ConcurrentOperationException {
         try {
-            if (!kubernetesClusterService.deleteKubernetesCluster(id)) {
+            if (!kubernetesClusterService.deleteKubernetesCluster(this)) {
                 throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, String.format("Failed to delete Kubernetes cluster ID: %d", getId()));
             }
             SuccessResponse response = new SuccessResponse(getCommandName());

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ListKubernetesClustersCmd.java

@@ -61,6 +61,10 @@ public class ListKubernetesClustersCmd extends BaseListProjectAndAccountResource
             " (a substring match is made against the parameter value, data for all matching Kubernetes clusters will be returned)")
     private String name;
 
+    @Parameter(name = ApiConstants.CLUSTER_TYPE, type = CommandType.STRING, since = "4.19.0",
+            description = "type of the cluster: CloudManaged, ExternalManaged")
+    private String clusterType;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -77,6 +81,10 @@ public class ListKubernetesClustersCmd extends BaseListProjectAndAccountResource
         return name;
     }
 
+    public String getClusterType() {
+        return clusterType;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/RemoveVirtualMachinesFromKubernetesClusterCmd.java

@@ -0,0 +1,103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.api.command.user.kubernetes.cluster;
+
+import com.cloud.kubernetes.cluster.KubernetesClusterService;
+import com.cloud.utils.exception.CloudRuntimeException;
+import org.apache.cloudstack.acl.RoleType;
+import org.apache.cloudstack.api.APICommand;
+import org.apache.cloudstack.api.ApiConstants;
+import org.apache.cloudstack.api.ApiErrorCode;
+import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.api.BaseListCmd;
+import org.apache.cloudstack.api.Parameter;
+import org.apache.cloudstack.api.ServerApiException;
+import org.apache.cloudstack.api.response.KubernetesClusterResponse;
+import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.cloudstack.api.response.RemoveVirtualMachinesFromKubernetesClusterResponse;
+import org.apache.cloudstack.api.response.UserVmResponse;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.log4j.Logger;
+
+import javax.inject.Inject;
+import java.util.List;
+
+@APICommand(name = "removeVirtualMachinesFromKubernetesCluster",
+        description = "Remove VMs from an ExternalManaged kubernetes cluster. Not applicable for CloudManaged kubernetes clusters.",
+        responseObject = RemoveVirtualMachinesFromKubernetesClusterResponse.class,
+        since = "4.19.0",
+        authorized = {RoleType.Admin, RoleType.ResourceAdmin, RoleType.DomainAdmin, RoleType.User})
+public class RemoveVirtualMachinesFromKubernetesClusterCmd extends BaseListCmd {
+    public static final Logger LOGGER = Logger.getLogger(RemoveVirtualMachinesFromKubernetesClusterCmd.class.getName());
+
+    @Inject
+    public KubernetesClusterService kubernetesClusterService;
+
+    /////////////////////////////////////////////////////
+    //////////////// API parameters /////////////////////
+    /////////////////////////////////////////////////////
+
+    @Parameter(name = ApiConstants.ID, type = BaseCmd.CommandType.UUID,
+            entityType = KubernetesClusterResponse.class,
+            required = true,
+            description = "the ID of the Kubernetes cluster")
+    private Long id;
+
+    @Parameter(name = ApiConstants.VIRTUAL_MACHINE_IDS, type = CommandType.LIST,
+            collectionType=CommandType.UUID,
+            entityType = UserVmResponse.class,
+            required = true,
+            description = "the IDs of the VMs to remove from the cluster")
+    private List<Long> vmIds;
+
+    /////////////////////////////////////////////////////
+    /////////////////// Accessors ///////////////////////
+    /////////////////////////////////////////////////////
+
+    public Long getId() {
+        return id;
+    }
+
+    public List<Long> getVmIds() {
+        return vmIds;
+    }
+
+    /////////////////////////////////////////////////////
+    /////////////// API Implementation///////////////////
+    /////////////////////////////////////////////////////
+
+    @Override
+    public long getEntityOwnerId() {
+        return CallContext.current().getCallingAccount().getId();
+    }
+
+    @Override
+    public void execute() throws ServerApiException {
+        try {
+            List<RemoveVirtualMachinesFromKubernetesClusterResponse> responseList = kubernetesClusterService.removeVmsFromCluster(this);
+            if (responseList.size() < 1) {
+                throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Provided VMs are not part of the CKS cluster");
+            }
+            ListResponse<RemoveVirtualMachinesFromKubernetesClusterResponse> listResponse = new ListResponse<>();
+            listResponse.setResponseName(getCommandName());
+            listResponse.setResponses(responseList);
+            setResponseObject(listResponse);
+        } catch (CloudRuntimeException e) {
+            throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, e.getMessage());
+        }
+    }
+}

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/ScaleKubernetesClusterCmd.java

@@ -43,7 +43,7 @@ import com.cloud.kubernetes.cluster.KubernetesClusterService;
 import com.cloud.utils.exception.CloudRuntimeException;
 
 @APICommand(name = "scaleKubernetesCluster",
-        description = "Scales a created, running or stopped Kubernetes cluster",
+        description = "Scales a created, running or stopped CloudManaged Kubernetes cluster",
         responseObject = KubernetesClusterResponse.class,
         responseView = ResponseObject.ResponseView.Restricted,
         entityType = {KubernetesCluster.class},

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StartKubernetesClusterCmd.java

@@ -36,7 +36,7 @@ import com.cloud.kubernetes.cluster.KubernetesClusterEventTypes;
 import com.cloud.kubernetes.cluster.KubernetesClusterService;
 import com.cloud.utils.exception.CloudRuntimeException;
 
-@APICommand(name = "startKubernetesCluster", description = "Starts a stopped Kubernetes cluster",
+@APICommand(name = "startKubernetesCluster", description = "Starts a stopped CloudManaged Kubernetes cluster",
         responseObject = KubernetesClusterResponse.class,
         responseView = ResponseObject.ResponseView.Restricted,
         entityType = {KubernetesCluster.class},
@@ -99,6 +99,10 @@ public class StartKubernetesClusterCmd extends BaseAsyncCmd {
         if (kubernetesCluster == null) {
             throw new ServerApiException(ApiErrorCode.PARAM_ERROR, "Given Kubernetes cluster was not found");
         }
+        if (!kubernetesClusterService.isCommandSupported(kubernetesCluster, getActualCommandName())) {
+            throw new ServerApiException(ApiErrorCode.PARAM_ERROR,
+                    String.format("Start kubernetes cluster is not supported for an externally managed cluster (%s)", kubernetesCluster.getName()));
+        }
         return kubernetesCluster;
     }
 

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/StopKubernetesClusterCmd.java

@@ -37,7 +37,7 @@ import com.cloud.kubernetes.cluster.KubernetesClusterEventTypes;
 import com.cloud.kubernetes.cluster.KubernetesClusterService;
 import com.cloud.utils.exception.CloudRuntimeException;
 
-@APICommand(name = "stopKubernetesCluster", description = "Stops a running Kubernetes cluster",
+@APICommand(name = "stopKubernetesCluster", description = "Stops a running CloudManaged Kubernetes cluster",
         responseObject = SuccessResponse.class,
         responseView = ResponseObject.ResponseView.Restricted,
         entityType = {KubernetesCluster.class},
@@ -95,7 +95,7 @@ public class StopKubernetesClusterCmd extends BaseAsyncCmd {
     @Override
     public void execute() throws ServerApiException, ConcurrentOperationException {
         try {
-            if (!kubernetesClusterService.stopKubernetesCluster(getId())) {
+            if (!kubernetesClusterService.stopKubernetesCluster(this)) {
                 throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, String.format("Failed to start Kubernetes cluster ID: %d", getId()));
             }
             final SuccessResponse response = new SuccessResponse(getCommandName());

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/command/user/kubernetes/cluster/UpgradeKubernetesClusterCmd.java

@@ -38,7 +38,7 @@ import com.cloud.kubernetes.cluster.KubernetesClusterEventTypes;
 import com.cloud.kubernetes.cluster.KubernetesClusterService;
 import com.cloud.utils.exception.CloudRuntimeException;
 
-@APICommand(name = "upgradeKubernetesCluster", description = "Upgrades a running Kubernetes cluster",
+@APICommand(name = "upgradeKubernetesCluster", description = "Upgrades a running CloudManaged Kubernetes cluster",
         responseObject = KubernetesClusterResponse.class,
         responseView = ResponseObject.ResponseView.Restricted,
         entityType = {KubernetesCluster.class},

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/KubernetesClusterResponse.java

@@ -159,6 +159,10 @@ public class KubernetesClusterResponse extends BaseResponseWithAnnotations imple
     @Param(description = "Maximum size of the cluster")
     private Long maxSize;
 
+    @SerializedName(ApiConstants.CLUSTER_TYPE)
+    @Param(description = "the type of the cluster")
+    private KubernetesCluster.ClusterType clusterType;
+
     @SerializedName(ApiConstants.CREATED)
     @Param(description = "the date when this Kubernetes cluster was created")
     private Date created;
@@ -386,4 +390,12 @@ public class KubernetesClusterResponse extends BaseResponseWithAnnotations imple
     public void setCreated(Date created) {
         this.created = created;
     }
+
+    public KubernetesCluster.ClusterType getClusterType() {
+        return clusterType;
+    }
+
+    public void setClusterType(KubernetesCluster.ClusterType clusterType) {
+        this.clusterType = clusterType;
+    }
 }

plugins/integrations/kubernetes-service/src/main/java/org/apache/cloudstack/api/response/RemoveVirtualMachinesFromKubernetesClusterResponse.java

@@ -0,0 +1,34 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.api.response;
+
+import com.cloud.serializer.Param;
+import com.google.gson.annotations.SerializedName;
+import org.apache.cloudstack.api.ApiConstants;
+
+public class RemoveVirtualMachinesFromKubernetesClusterResponse extends SuccessResponse {
+    @SerializedName(ApiConstants.ID)
+    @Param(description = "the id of the Kubernetes cluster")
+    private String vmId;
+
+    public RemoveVirtualMachinesFromKubernetesClusterResponse() {
+    }
+
+    public void setVmId(String vmId) {
+        this.vmId = vmId;
+    }
+}

plugins/integrations/kubernetes-service/src/test/java/com/cloud/kubernetes/cluster/KubernetesClusterManagerImplTest.java

@@ -1,25 +1,52 @@
-// Licensed to the Apache Software Foundation (ASF) under one
+/*
-// or more contributor license agreements.  See the NOTICE file
+ * Licensed to the Apache Software Foundation (ASF) under one
-// distributed with this work for additional information
+ * or more contributor license agreements.  See the NOTICE file
-// regarding copyright ownership.  The ASF licenses this file
+ * distributed with this work for additional information
-// to you under the Apache License, Version 2.0 (the
+ * regarding copyright ownership.  The ASF licenses this file
-// "License"); you may not use this file except in compliance
+ * to you under the Apache License, Version 2.0 (the
-// with the License.  You may obtain a copy of the License at
+ * "License"); you may not use this file except in compliance
-//
+ * with the License.  You may obtain a copy of the License at
-//   http://www.apache.org/licenses/LICENSE-2.0
+ *
-//
+ *   http://www.apache.org/licenses/LICENSE-2.0
-// Unless required by applicable law or agreed to in writing,
+ *
-// software distributed under the License is distributed on an
+ * Unless required by applicable law or agreed to in writing,
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * software distributed under the License is distributed on an
-// KIND, either express or implied.  See the License for the
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// specific language governing permissions and limitations
+ * KIND, either express or implied.  See the License for the
-// under the License.
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
 package com.cloud.kubernetes.cluster;
 
-
+import com.cloud.api.query.dao.TemplateJoinDao;
-import java.util.ArrayList;
+import com.cloud.api.query.vo.TemplateJoinVO;
-import java.util.List;
+import com.cloud.dc.DataCenter;
-
+import com.cloud.exception.InvalidParameterValueException;
+import com.cloud.exception.PermissionDeniedException;
+import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterActionWorker;
+import com.cloud.kubernetes.cluster.dao.KubernetesClusterDao;
+import com.cloud.kubernetes.cluster.dao.KubernetesClusterVmMapDao;
+import com.cloud.network.Network;
+import com.cloud.network.dao.FirewallRulesDao;
+import com.cloud.network.rules.FirewallRule;
+import com.cloud.network.rules.FirewallRuleVO;
+import com.cloud.network.vpc.NetworkACL;
+import com.cloud.storage.VMTemplateVO;
+import com.cloud.storage.dao.VMTemplateDao;
+import com.cloud.user.Account;
+import com.cloud.user.AccountManager;
+import com.cloud.user.User;
+import com.cloud.vm.VMInstanceVO;
+import com.cloud.vm.dao.VMInstanceDao;
+import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.AddVirtualMachinesToKubernetesClusterCmd;
+import org.apache.cloudstack.api.command.user.kubernetes.cluster.RemoveVirtualMachinesFromKubernetesClusterCmd;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
@@ -28,19 +55,11 @@ import org.mockito.Mockito;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import com.cloud.api.query.dao.TemplateJoinDao;
+import java.lang.reflect.Field;
-import com.cloud.api.query.vo.TemplateJoinVO;
+import java.util.ArrayList;
-import com.cloud.dc.DataCenter;
+import java.util.Arrays;
-import com.cloud.exception.InvalidParameterValueException;
+import java.util.Collections;
-import com.cloud.exception.PermissionDeniedException;
+import java.util.List;
-import com.cloud.kubernetes.cluster.actionworkers.KubernetesClusterActionWorker;
-import com.cloud.network.Network;
-import com.cloud.network.dao.FirewallRulesDao;
-import com.cloud.network.rules.FirewallRule;
-import com.cloud.network.rules.FirewallRuleVO;
-import com.cloud.network.vpc.NetworkACL;
-import com.cloud.storage.VMTemplateVO;
-import com.cloud.storage.dao.VMTemplateDao;
 
 @RunWith(MockitoJUnitRunner.class)
 public class KubernetesClusterManagerImplTest {
@@ -54,6 +73,18 @@ public class KubernetesClusterManagerImplTest {
     @Mock
     TemplateJoinDao templateJoinDao;
 
+    @Mock
+    KubernetesClusterDao kubernetesClusterDao;
+
+    @Mock
+    KubernetesClusterVmMapDao kubernetesClusterVmMapDao;
+
+    @Mock
+    VMInstanceDao vmInstanceDao;
+
+    @Mock
+    private AccountManager accountManager;
+
     @Spy
     @InjectMocks
     KubernetesClusterManagerImpl kubernetesClusterManager;
@@ -171,7 +202,7 @@ public class KubernetesClusterManagerImplTest {
     }
 
     @Test
-    public void testValidateKubernetesClusterScaleSizeDownsacaleNoError() {
+    public void testValidateKubernetesClusterScaleSizeDownscaleNoError() {
         KubernetesClusterVO clusterVO = Mockito.mock(KubernetesClusterVO.class);
         Mockito.when(clusterVO.getState()).thenReturn(KubernetesCluster.State.Running);
         Mockito.when(clusterVO.getControlNodeCount()).thenReturn(1L);
@@ -209,5 +240,56 @@ public class KubernetesClusterManagerImplTest {
         Mockito.when(templateDao.findById(Mockito.anyLong())).thenReturn(Mockito.mock(VMTemplateVO.class));
         Mockito.when(templateJoinDao.newTemplateView(Mockito.any(VMTemplateVO.class), Mockito.anyLong(), Mockito.anyBoolean())).thenReturn(List.of(Mockito.mock(TemplateJoinVO.class)));
         kubernetesClusterManager.validateKubernetesClusterScaleSize(clusterVO, 4L, 10, Mockito.mock(DataCenter.class));
+
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        CallContext.register(Mockito.mock(User.class), Mockito.mock(Account.class));
+        overrideDefaultConfigValue(KubernetesClusterService.KubernetesServiceEnabled, "_defaultValue", "true");
+        Mockito.doNothing().when(accountManager).checkAccess(
+                Mockito.any(Account.class), Mockito.any(), Mockito.anyBoolean(), Mockito.any());
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        CallContext.unregister();
+    }
+
+    private void overrideDefaultConfigValue(final ConfigKey configKey, final String name, final Object o) throws IllegalAccessException, NoSuchFieldException {
+        Field f = ConfigKey.class.getDeclaredField(name);
+        f.setAccessible(true);
+        f.set(configKey, o);
+    }
+
+    @Test
+    public void addVmsToCluster() {
+        KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class);
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        AddVirtualMachinesToKubernetesClusterCmd cmd = Mockito.mock(AddVirtualMachinesToKubernetesClusterCmd.class);
+        List<Long> vmIds = Arrays.asList(1L, 2L, 3L);
+
+        Mockito.when(cmd.getId()).thenReturn(1L);
+        Mockito.when(cmd.getVmIds()).thenReturn(vmIds);
+        Mockito.when(cmd.getActualCommandName()).thenReturn(BaseCmd.getCommandNameByClass(RemoveVirtualMachinesFromKubernetesClusterCmd.class));
+        Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged);
+        Mockito.when(vmInstanceDao.findById(Mockito.anyLong())).thenReturn(vm);
+        Mockito.when(kubernetesClusterDao.findById(Mockito.anyLong())).thenReturn(cluster);
+        Mockito.when(kubernetesClusterVmMapDao.listByClusterIdAndVmIdsIn(1L, vmIds)).thenReturn(Collections.emptyList());
+        Assert.assertTrue(kubernetesClusterManager.addVmsToCluster(cmd));
+    }
+
+    @Test
+    public void removeVmsFromCluster() {
+        KubernetesClusterVO cluster = Mockito.mock(KubernetesClusterVO.class);
+        RemoveVirtualMachinesFromKubernetesClusterCmd cmd = Mockito.mock(RemoveVirtualMachinesFromKubernetesClusterCmd.class);
+        List<Long> vmIds = Arrays.asList(1L, 2L, 3L);
+
+        Mockito.when(cmd.getId()).thenReturn(1L);
+        Mockito.when(cmd.getVmIds()).thenReturn(vmIds);
+        Mockito.when(cmd.getActualCommandName()).thenReturn(BaseCmd.getCommandNameByClass(RemoveVirtualMachinesFromKubernetesClusterCmd.class));
+        Mockito.when(cluster.getClusterType()).thenReturn(KubernetesCluster.ClusterType.ExternalManaged);
+        Mockito.when(kubernetesClusterDao.findById(Mockito.anyLong())).thenReturn(cluster);
+        Assert.assertTrue(kubernetesClusterManager.removeVmsFromCluster(cmd).size() > 0);
     }
 }

test/integration/smoke/test_kubernetes_clusters.py

@@ -33,7 +33,9 @@ from marvin.cloudstackAPI import (listInfrastructure,
                                   scaleKubernetesCluster,
                                   getKubernetesClusterConfig,
                                   destroyVirtualMachine,
-                                  deleteNetwork)
+                                  deleteNetwork,
+                                  addVirtualMachinesToKubernetesCluster,
+                                  removeVirtualMachinesFromKubernetesCluster)
 from marvin.cloudstackException import CloudstackAPIException
 from marvin.codes import PASS, FAILED
 from marvin.lib.base import (Template,
@@ -46,12 +48,14 @@ from marvin.lib.base import (Template,
                              VpcOffering,
                              VPC,
                              NetworkACLList,
-                             NetworkACL)
+                             NetworkACL,
+                             VirtualMachine)
 from marvin.lib.utils import (cleanup_resources,
                               validateList,
                               random_gen)
 from marvin.lib.common import (get_zone,
-                               get_domain)
+                               get_domain,
+                               get_template)
 from marvin.sshClient import SshClient
 from nose.plugins.attrib import attr
 from marvin.lib.decoratorGenerators import skipTestIf
@@ -86,6 +90,7 @@ class TestKubernetesCluster(cloudstackTestCase):
         cls._cleanup = []
         cls.kubernetes_version_ids = []
         cls.vpcAllowAllAclDetailsMap = {}
+        cls.initial_configuration_cks_enabled = None
 
         if cls.hypervisorNotSupported == False:
             cls.endpoint_url = Configurations.list(cls.apiclient, name="endpoint.url")[0].value
@@ -610,7 +615,82 @@ class TestKubernetesCluster(cloudstackTestCase):
         k8s_cluster = None
         return
 
-    def createKubernetesCluster(self, name, version_id, size=1, control_nodes=1):
+    @attr(tags=["advanced", "smoke"], required_hardware="false")
+    def test_11_test_unmanaged_cluster_lifecycle(self):
+        """Test all operations on unmanaged Kubernetes cluster
+
+        # Validate the following:
+        # 1. createKubernetesCluster should return valid info for new cluster
+        # 2. The Cloud Database contains the valid information
+        # 3. stopKubernetesCluster doesn't work
+        # 4. startKubernetesCluster doesn't work
+        # 5. upgradeKubernetesCluster doesn't work
+        # 6. Adding & removing vm from cluster works
+        # 7. deleteKubernetesCluster should delete an existing HA Kubernetes cluster
+        """
+        cluster = self.createKubernetesCluster("test-unmanaged-cluster", None,
+                                               cluster_type="ExternalManaged")
+        self.verifyKubernetesClusterState(cluster, 'Running')
+        self.debug("Stopping unmanaged Kubernetes cluster with ID: %s" % cluster.id)
+        try:
+            self.stopKubernetesCluster(cluster.id)
+            self.fail("Should not be able to stop unmanaged cluster")
+        except Exception as e:
+            self.debug("Expected exception: %s" % e)
+
+        self.debug("Starting unmanaged Kubernetes cluster with ID: %s" % cluster.id)
+        try:
+            self.startKubernetesCluster(cluster.id)
+            self.fail("Should not be able to start unmanaged cluster")
+        except Exception as e:
+            self.debug("Expected exception: %s" % e)
+
+        self.debug("Upgrading unmanaged Kubernetes cluster with ID: %s" % cluster.id)
+        try:
+            self.upgradeKubernetesCluster(cluster.id, self.kubernetes_version_1_24_0.id)
+            self.fail("Should not be able to upgrade unmanaged cluster")
+        except Exception as e:
+            self.debug("Expected exception: %s" % e)
+
+        template = get_template(self.apiclient,
+                                    self.zone.id,
+                                    self.services["ostype"])
+
+        self.services["virtual_machine"]["template"] = template.id
+        virtualMachine = VirtualMachine.create(self.apiclient, self.services["virtual_machine"], zoneid=self.zone.id,
+                                                            accountid=self.account.name, domainid=self.account.domainid,
+                                                            serviceofferingid=self.cks_service_offering.id)
+        self.debug("Adding VM %s to unmanaged Kubernetes cluster with ID: %s" % (virtualMachine.id, cluster.id))
+        self.addVirtualMachinesToKubernetesCluster(cluster.id, [virtualMachine.id])
+        cluster = self.listKubernetesCluster(cluster.id)
+        self.assertEqual(virtualMachine.id, cluster.virtualmachines[0].id, "VM should be part of the kubernetes cluster")
+        self.assertEqual(1, len(cluster.virtualmachines), "Only one VM should be part of the kubernetes cluster")
+
+        self.debug("Removing VM %s from unmanaged Kubernetes cluster with ID: %s" % (virtualMachine.id, cluster.id))
+        self.removeVirtualMachinesFromKubernetesCluster(cluster.id, [virtualMachine.id])
+        cluster = self.listKubernetesCluster(cluster.id)
+        self.assertEqual(0, len(cluster.virtualmachines), "No VM should be part of the kubernetes cluster")
+
+        self.debug("Deleting unmanaged Kubernetes cluster with ID: %s" % cluster.id)
+        self.deleteKubernetesClusterAndVerify(cluster.id)
+        return
+
+    def addVirtualMachinesToKubernetesCluster(self, cluster_id, vm_list):
+        cmd = addVirtualMachinesToKubernetesCluster.addVirtualMachinesToKubernetesClusterCmd()
+        cmd.id = cluster_id
+        cmd.virtualmachineids = vm_list
+
+        return self.apiclient.addVirtualMachinesToKubernetesCluster(cmd)
+
+    def removeVirtualMachinesFromKubernetesCluster(self, cluster_id, vm_list):
+        cmd = removeVirtualMachinesFromKubernetesCluster.removeVirtualMachinesFromKubernetesClusterCmd()
+        cmd.id = cluster_id
+        cmd.virtualmachineids = vm_list
+
+        return self.apiclient.removeVirtualMachinesFromKubernetesCluster(cmd)
+
+
+    def createKubernetesCluster(self, name, version_id, size=1, control_nodes=1, cluster_type='CloudManaged'):
         createKubernetesClusterCmd = createKubernetesCluster.createKubernetesClusterCmd()
         createKubernetesClusterCmd.name = name
         createKubernetesClusterCmd.description = name + "-description"
@@ -622,11 +702,10 @@ class TestKubernetesCluster(cloudstackTestCase):
         createKubernetesClusterCmd.noderootdisksize = 10
         createKubernetesClusterCmd.account = self.account.name
         createKubernetesClusterCmd.domainid = self.domain.id
+        createKubernetesClusterCmd.clustertype = cluster_type
         if self.default_network:
             createKubernetesClusterCmd.networkid = self.default_network.id
         clusterResponse = self.apiclient.createKubernetesCluster(createKubernetesClusterCmd)
-        if not clusterResponse:
-            self.cleanup.append(clusterResponse)
         return clusterResponse
 
     def startKubernetesCluster(self, cluster_id):

ui/public/locales/en.json

@@ -447,6 +447,7 @@
 "label.clear.list": "Clear list",
 "label.clear.notification": "Clear notification",
 "label.close": "Close",
+"label.cloud.managed": "CloudManaged",
 "label.cloudian.storage": "Cloudian storage",
 "label.cluster": "Cluster",
 "label.cluster.name": "Cluster name",
@@ -834,6 +835,7 @@
 "label.expunged": "Expunged",
 "label.expunging": "Expunging",
 "label.export.rules": "Export Rules",
+"label.external.managed": "ExternalManaged",
 "label.external.link": "External link",
 "label.externalid": "External Id",
 "label.externalloadbalanceripaddress": "External load balancer IP address.",
@@ -2057,6 +2059,7 @@
 "label.unit": "Usage unit",
 "label.unknown": "Unknown",
 "label.unlimited": "Unlimited",
+"label.unmanaged": "Unmanaged",
 "label.unmanage.instance": "Unmanage instance",
 "label.unmanaged.instance": "Unmanaged instance",
 "label.unmanaged.instances": "Unmanaged instances",

ui/src/components/view/ListView.vue

@@ -92,6 +92,9 @@
           </span>
         </span>
       </template>
+      <template v-if="record.clustertype === 'ExternalManaged' && $route.path.split('/')[1] === 'kubernetes' && ['cpunumber', 'memory', 'size'].includes(column.key)">
+        <span>{{ text <= 0 ? 'N/A' : text }}</span>
+      </template>
       <template v-if="column.key === 'templatetype'">
         <router-link :to="{ path: $route.path + '/' + record.templatetype }">{{ text }}</router-link>
       </template>

ui/src/config/section/compute.js

@@ -447,7 +447,7 @@ export default {
       docHelp: 'plugins/cloudstack-kubernetes-service.html',
       permission: ['listKubernetesClusters'],
       columns: (store) => {
-        var fields = ['name', 'state', 'size', 'cpunumber', 'memory']
+        var fields = ['name', 'state', 'clustertype', 'size', 'cpunumber', 'memory']
         if (['Admin', 'DomainAdmin'].includes(store.userInfo.roletype)) {
           fields.push('account')
         }
@@ -457,7 +457,11 @@ export default {
         fields.push('zonename')
         return fields
       },
-      details: ['name', 'description', 'zonename', 'kubernetesversionname', 'autoscalingenabled', 'minsize', 'maxsize', 'size', 'controlnodes', 'cpunumber', 'memory', 'keypair', 'associatednetworkname', 'account', 'domain', 'zonename', 'created'],
+      filters: () => {
+        const filters = ['cloud.managed', 'external.managed']
+        return filters
+      },
+      details: ['name', 'description', 'zonename', 'kubernetesversionname', 'autoscalingenabled', 'minsize', 'maxsize', 'size', 'controlnodes', 'cpunumber', 'memory', 'keypair', 'associatednetworkname', 'account', 'domain', 'zonename', 'clustertype', 'created'],
       tabs: [{
         name: 'k8s',
         component: shallowRef(defineAsyncComponent(() => import('@/views/compute/KubernetesServiceTab.vue')))
@@ -480,7 +484,7 @@ export default {
           message: 'message.kubernetes.cluster.start',
           docHelp: 'plugins/cloudstack-kubernetes-service.html#starting-a-stopped-kubernetes-cluster',
           dataView: true,
-          show: (record) => { return ['Stopped'].includes(record.state) },
+          show: (record) => { return ['Stopped'].includes(record.state) && record.clustertype === 'CloudManaged' },
           groupAction: true,
           popup: true,
           groupMap: (selection) => { return selection.map(x => { return { id: x } }) }
@@ -492,7 +496,7 @@ export default {
           message: 'message.kubernetes.cluster.stop',
           docHelp: 'plugins/cloudstack-kubernetes-service.html#stopping-kubernetes-cluster',
           dataView: true,
-          show: (record) => { return !['Stopped', 'Destroyed', 'Destroying'].includes(record.state) },
+          show: (record) => { return !['Stopped', 'Destroyed', 'Destroying'].includes(record.state) && record.clustertype === 'CloudManaged' },
           groupAction: true,
           popup: true,
           groupMap: (selection) => { return selection.map(x => { return { id: x } }) }
@@ -504,7 +508,7 @@ export default {
           message: 'message.kubernetes.cluster.scale',
           docHelp: 'plugins/cloudstack-kubernetes-service.html#scaling-kubernetes-cluster',
           dataView: true,
-          show: (record) => { return ['Created', 'Running', 'Stopped'].includes(record.state) },
+          show: (record) => { return ['Created', 'Running', 'Stopped'].includes(record.state) && record.clustertype === 'CloudManaged' },
           popup: true,
           component: shallowRef(defineAsyncComponent(() => import('@/views/compute/ScaleKubernetesCluster.vue')))
         },
@@ -515,7 +519,7 @@ export default {
           message: 'message.kubernetes.cluster.upgrade',
           docHelp: 'plugins/cloudstack-kubernetes-service.html#upgrading-kubernetes-cluster',
           dataView: true,
-          show: (record) => { return ['Created', 'Running'].includes(record.state) },
+          show: (record) => { return ['Created', 'Running'].includes(record.state) && record.clustertype === 'CloudManaged' },
           popup: true,
           component: shallowRef(defineAsyncComponent(() => import('@/views/compute/UpgradeKubernetesCluster.vue')))
         },
@@ -529,7 +533,11 @@ export default {
           show: (record) => { return !['Destroyed', 'Destroying'].includes(record.state) },
           groupAction: true,
           popup: true,
-          groupMap: (selection) => { return selection.map(x => { return { id: x } }) }
+          args: (record, store, group) => {
+            return (['Admin'].includes(store.userInfo.roletype) || store.features.allowuserexpungerecovervm)
+              ? ['cleanup', 'expunge'] : ['cleanup']
+          },
+          groupMap: (selection, values) => { return selection.map(x => { return { id: x, expunge: values.expunge, cleanup: values.cleanup } }) }
         }
       ]
     },

ui/src/views/AutogenView.vue

@@ -64,7 +64,7 @@
                     <template #suffixIcon><filter-outlined class="ant-select-suffix" /></template>
                     <a-select-option
                       v-if="['Admin', 'DomainAdmin'].includes($store.getters.userInfo.roletype) &&
-                      ['vm', 'iso', 'template', 'pod', 'cluster', 'host', 'systemvm', 'router', 'storagepool'].includes($route.name) ||
+                      ['vm', 'iso', 'template', 'pod', 'cluster', 'host', 'systemvm', 'router', 'storagepool', 'kubernetes'].includes($route.name) ||
                       ['account'].includes($route.name)"
                       key="all"
                       :label="$t('label.all')">
@@ -676,7 +676,7 @@ export default {
         return this.$route.query.filter
       }
       const routeName = this.$route.name
-      if ((this.projectView && routeName === 'vm') || (['Admin', 'DomainAdmin'].includes(this.$store.getters.userInfo.roletype) && ['vm', 'iso', 'template', 'pod', 'cluster', 'host', 'systemvm', 'router', 'storagepool'].includes(routeName)) || ['account', 'guestnetwork', 'guestvlans', 'guestos', 'guestoshypervisormapping'].includes(routeName)) {
+      if ((this.projectView && routeName === 'vm') || (['Admin', 'DomainAdmin'].includes(this.$store.getters.userInfo.roletype) && ['vm', 'iso', 'template', 'pod', 'cluster', 'host', 'systemvm', 'router', 'storagepool'].includes(routeName)) || ['account', 'guestnetwork', 'guestvlans', 'guestos', 'guestoshypervisormapping', 'kubernetes'].includes(routeName)) {
         return 'all'
       }
       if (['publicip'].includes(routeName)) {
@@ -1657,6 +1657,12 @@ export default {
         } else {
           query.hypervisor = filter
         }
+      } else if (this.$route.name === 'kubernetes') {
+        if (filter === 'all') {
+          delete query.clustertype
+        } else {
+          query.clustertype = filter === 'cloud.managed' ? 'CloudManaged' : 'ExternalManaged'
+        }
       }
       query.filter = filter
       query.page = '1'

ui/src/views/compute/CreateKubernetesCluster.vue

@@ -459,7 +459,8 @@ export default {
           zoneid: this.zones[values.zoneid].id,
           kubernetesversionid: this.kubernetesVersions[values.kubernetesversionid].id,
           serviceofferingid: this.serviceOfferings[values.serviceofferingid].id,
-          size: values.size
+          size: values.size,
+          clustertype: 'CloudManaged'
         }
         if (this.isValidValueForKey(values, 'noderootdisksize') && values.noderootdisksize > 0) {
           params.noderootdisksize = values.noderootdisksize

ui/src/views/compute/KubernetesServiceTab.vue

@@ -25,7 +25,7 @@
       <a-tab-pane :tab="$t('label.details')" key="details">
         <DetailsTab :resource="resource" :loading="loading" />
       </a-tab-pane>
-      <a-tab-pane :tab="$t('label.access')" key="access">
+      <a-tab-pane v-if="resource.clustertype == 'CloudManaged'" :tab="$t('label.access')" key="access">
         <a-card :title="$t('label.kubeconfig.cluster')" :loading="versionLoading">
           <div v-if="clusterConfig !== ''">
             <a-textarea :value="clusterConfig" :rows="5" readonly />
@@ -240,6 +240,9 @@ export default {
     if (!isAdmin()) {
       this.vmColumns = this.vmColumns.filter(x => x.dataIndex !== 'instancename')
     }
+    if (this.resource.clustertype === 'ExternalManaged') {
+      this.vmColumns = this.vmColumns.filter(x => x.dataIndex !== 'port')
+    }
     this.handleFetchData()
     const self = this
     window.addEventListener('popstate', function () {