[AKS] az aks stop: Add warning when private link cluster is stopped

src/azure-cli/azure/cli/command_modules/acs/_help.py

@@ -1734,6 +1734,16 @@
     crafted: true
 """
 
+helps['aks stop'] = """
+    type: command
+    short-summary: Stop a managed cluster.
+    long-summary: This can only be performed on Azure Virtual Machine Scale set backed clusters. Stopping a
+        cluster stops the control plane and agent nodes entirely, while maintaining all object and
+        cluster state. A cluster does not accrue charges while it is stopped. See `stopping a
+        cluster <https://docs.microsoft.com/azure/aks/start-stop-cluster>`_ for more details about
+        stopping a cluster.
+"""
+
 helps['aks update-credentials'] = """
 type: command
 short-summary: Update credentials for a managed Kubernetes cluster, like service principal.

src/azure-cli/azure/cli/command_modules/acs/_helpers.py

@@ -98,6 +98,28 @@ def check_is_private_cluster(mc: ManagedCluster) -> bool:
     return False
 
 
+def check_is_apiserver_vnet_integration_cluster(mc: ManagedCluster) -> bool:
+    """Check `mc` object to determine whether apiserver vnet integration is enabled.
+
+    Note: enableVnetIntegration is still in preview api so we use additional_properties here
+
+    :return: bool
+    """
+    if mc and mc.api_server_access_profile:
+        additional_properties = mc.api_server_access_profile.additional_properties
+        if 'enableVnetIntegration' in additional_properties:
+            return additional_properties['enableVnetIntegration']
+        return False
+    return False
+
+
+def check_is_private_link_cluster(mc: ManagedCluster) -> bool:
+    """Check `mc` object to determine whether private link cluster is enabled.
+    :return: bool
+    """
+    return check_is_private_cluster(mc) and not check_is_apiserver_vnet_integration_cluster(mc)
+
+
 def check_is_managed_aad_cluster(mc: ManagedCluster) -> bool:
     """Check `mc` object to determine whether managed aad is enabled.
 

src/azure-cli/azure/cli/command_modules/acs/commands.py

@@ -81,7 +81,7 @@ def load_command_table(self, _):
         g.custom_command('rotate-certs', 'aks_rotate_certs', supports_no_wait=True,
                          confirmation='Kubernetes will be unavailable during certificate rotation process.\n' +
                          'Are you sure you want to perform this operation?')
-        g.command('stop', 'begin_stop', supports_no_wait=True, min_api='2020-09-01')
+        g.custom_command('stop', 'aks_stop', supports_no_wait=True, min_api='2020-09-01')
         g.command('start', 'begin_start', supports_no_wait=True, min_api='2020-09-01')
         g.wait_command('wait')
         g.custom_command('use-dev-spaces', 'aks_use_dev_spaces', deprecate_info=g.deprecate())

src/azure-cli/azure/cli/command_modules/acs/custom.py

@@ -62,7 +62,7 @@
     DecoratorEarlyExitException,
 )
 
-from azure.cli.command_modules.acs._helpers import get_snapshot_by_snapshot_id
+from azure.cli.command_modules.acs._helpers import get_snapshot_by_snapshot_id, check_is_private_link_cluster
 from azure.cli.command_modules.acs._resourcegroup import get_rg_location
 from azure.cli.command_modules.acs._validators import extract_comma_separated_string
 from azure.cli.command_modules.acs.addonconfiguration import (
@@ -900,6 +900,16 @@ def aks_show(cmd, client, resource_group_name, name):
     return _remove_nulls([mc])[0]
 
 
+def aks_stop(cmd, client, resource_group_name, name, no_wait=False):
+    instance = client.get(resource_group_name, name)
+    # print warning when stopping a private link cluster
+    if check_is_private_link_cluster(instance):
+        logger.warning('Your private cluster apiserver IP might get changed when it\'s stopped and started.\n'
+                       'Any user provisioned private endpoints linked to this private cluster will need to be deleted and created again. '
+                       'Any user managed DNS record also needs to be updated with the new IP.')
+    return sdk_no_wait(no_wait, client.begin_stop, resource_group_name, name)
+
+
 def aks_list(cmd, client, resource_group_name=None):
     if resource_group_name:
         managed_clusters = client.list_by_resource_group(resource_group_name)

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_aks_commands.py

@@ -5100,6 +5100,27 @@ def test_aks_stop_and_start(self, resource_group, resource_group_location):
         start_cmd = 'aks start --resource-group={resource_group} --name={name}'
         self.cmd(start_cmd)
 
+    @AllowLargeResponse()
+    @AKSCustomResourceGroupPreparer(random_name_length=17, name_prefix='clitest', location='westus2')
+    def test_aks_stop_and_start_private_cluster(self, resource_group, resource_group_location):
+        aks_name = self.create_random_name('cliakstest', 16)
+        self.kwargs.update({
+            'resource_group': resource_group,
+            'name': aks_name,
+            'ssh_key_value': self.generate_ssh_keys(),
+        })
+
+        create_cmd = 'aks create --resource-group={resource_group} --name={name} --ssh-key-value={ssh_key_value} --enable-private-cluster'
+        self.cmd(create_cmd, checks=[
+            self.check('provisioningState', 'Succeeded'),
+        ])
+
+        stop_cmd = 'aks stop --resource-group={resource_group} --name={name}'
+        self.cmd(stop_cmd)
+
+        start_cmd = 'aks start --resource-group={resource_group} --name={name}'
+        self.cmd(start_cmd)
+
     @AllowLargeResponse()
     @AKSCustomResourceGroupPreparer(random_name_length=17, name_prefix='clitest', location='westus2')
     def test_aks_abort(self, resource_group, resource_group_location):

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_custom.py

@@ -19,12 +19,17 @@
 from azure.cli.command_modules.acs.custom import (
     _get_command_context,
     _update_addons,
+    aks_stop,
     k8s_install_kubectl,
     k8s_install_kubelogin,
     merge_kubernetes_configurations,
 )
+from azure.cli.command_modules.acs.managed_cluster_decorator import (
+    AKSManagedClusterModels,
+)
 from azure.cli.command_modules.acs.tests.latest.mocks import (
     MockCLI,
+    MockClient,
     MockCmd,
     MockUrlretrieveUrlValidator,
 )
@@ -33,6 +38,7 @@
     get_test_data_file_path,
 )
 from azure.cli.core.util import CLIError
+from azure.cli.core.profiles import ResourceType
 from azure.mgmt.containerservice.v2020_03_01.models import (
     ManagedClusterAddonProfile,
 )
@@ -677,6 +683,37 @@ def test_k8s_install_kubelogin_with_custom_source_url(self, logger_mock, mock_ur
         finally:
             shutil.rmtree(temp_dir)
 
+class TestAKSCommand(unittest.TestCase):
+    def setUp(self):
+        self.cli_ctx = MockCLI()
+        self.cmd = MockCmd(self.cli_ctx)
+        self.models = AKSManagedClusterModels(self.cmd, ResourceType.MGMT_CONTAINERSERVICE)
+        self.client = MockClient()
+
+    def test_aks_stop(self):
+        # public cluster: call begin_stop
+        mc_1 = self.models.ManagedCluster(location="test_location")
+        self.client.get = mock.Mock(
+            return_value=mc_1
+        )
+        self.client.begin_stop = mock.Mock(
+            return_value=None
+        )
+        self.assertEqual(aks_stop(self.cmd, self.client, "rg", "name"), None)
+
+        # private cluster: call begin_stop
+        mc_2 = self.models.ManagedCluster(location="test_location")
+        api_server_access_profile = self.models.ManagedClusterAPIServerAccessProfile()
+        api_server_access_profile.enable_private_cluster = True
+        mc_2.api_server_access_profile = api_server_access_profile
+        self.client.get = mock.Mock(
+            return_value=mc_2
+        )
+        self.client.begin_stop = mock.Mock(
+            return_value=None
+        )
+        self.assertEqual(aks_stop(self.cmd, self.client, "rg", "name", False), None)
+
 
 class TestRunCommand(unittest.TestCase):
     def test_get_command_context_invalid_file(self):

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_helpers.py

@@ -7,9 +7,11 @@
 from unittest.mock import Mock, patch
 
 from azure.cli.command_modules.acs._helpers import (
+    check_is_apiserver_vnet_integration_cluster,
     check_is_managed_aad_cluster,
     check_is_msi_cluster,
     check_is_private_cluster,
+    check_is_private_link_cluster,
     format_parameter_name_to_option_name,
     get_property_from_dict_or_object,
     get_snapshot,
@@ -119,6 +121,73 @@ def test_check_is_private_cluster(self):
         )
         self.assertEqual(check_is_private_cluster(mc_4), False)
 
+    def test_check_is_apiserver_vnet_integration_cluster(self):
+        self.assertEqual(check_is_apiserver_vnet_integration_cluster(None), False)
+
+        mc_1 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(),
+        )
+        mc_1.api_server_access_profile.additional_properties={'enableVnetIntegration': True}
+        self.assertEqual(check_is_apiserver_vnet_integration_cluster(mc_1), True)
+
+        mc_2 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(),
+        )
+        mc_2.api_server_access_profile.additional_properties={'enableVnetIntegration': False}
+        self.assertEqual(check_is_apiserver_vnet_integration_cluster(mc_2), False)
+
+        mc_3 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(),
+        )
+        self.assertEqual(check_is_apiserver_vnet_integration_cluster(mc_3), False)
+
+        mc_4 = self.models.ManagedCluster(
+            location="test_location",
+        )
+        self.assertEqual(check_is_apiserver_vnet_integration_cluster(mc_4), False)
+
+    def test_check_is_private_link_cluster(self):
+        self.assertEqual(check_is_private_link_cluster(None), False)
+
+        mc_1 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(
+                enable_private_cluster=True,
+            ),
+        )
+        self.assertEqual(check_is_private_link_cluster(mc_1), True)
+
+        mc_2 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(
+                enable_private_cluster=False,
+            ),
+        )
+        self.assertEqual(check_is_private_link_cluster(mc_2), False)
+
+        mc_3 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(
+                enable_private_cluster=True,
+            ),
+        )
+        mc_3.api_server_access_profile.additional_properties={'enableVnetIntegration': True}
+        self.assertEqual(check_is_private_link_cluster(mc_3), False)
+
+        mc_4 = self.models.ManagedCluster(
+            location="test_location",
+            api_server_access_profile=self.models.ManagedClusterAPIServerAccessProfile(),
+        )
+        self.assertEqual(check_is_private_link_cluster(mc_4), False)
+
+        mc_5 = self.models.ManagedCluster(
+            location="test_location",
+        )
+        self.assertEqual(check_is_private_link_cluster(mc_5), False)
+
     def test_check_is_managed_aad_cluster(self):
         self.assertEqual(check_is_managed_aad_cluster(None), False)