[Container App] Add ContainerApp Health Check commands

src/containerapp/azext_containerapp/_help.py

@@ -2054,3 +2054,21 @@
           az containerapp job registry set -n my-containerapp-job -g MyResourceGroup \\
               --server MyContainerappJobRegistry.azurecr.io --identity system-environment
 """
+
+helps['containerapp diagnose dns resolve'] = """
+    type: command
+    short-summary: Diagnose issues with a container app environment.
+    examples:
+    - name: Check DNS resolution for a container app environment.
+      text: |
+          az containerapp diagnose dns -n MyContainerappEnvironment -g MyResourceGroup --domain www.microsoft.com
+"""
+
+helps['containerapp diagnose healthcheck show'] = """
+    type: command
+    short-summary: Show health check status for a container app environment.
+    examples:
+    - name: Show health check status for a container app environment.
+      text: |
+          az containerapp diagnose healthcheck show -n MyContainerappEnvironment -g MyResourceGroup
+"""

src/containerapp/azext_containerapp/_params.py

@@ -384,6 +384,22 @@ def load_arguments(self, _):
     with self.argument_context('containerapp job registry') as c:
         c.argument('identity', help="The managed identity with which to authenticate to the Azure Container Registry (instead of username/password). Use 'system' for a system-defined identity, Use 'system-environment' for an environment level system-defined identity or a resource id for a user-defined environment/containerapp level identity. The managed identity should have been assigned acrpull permissions on the ACR before deployment (use 'az role assignment create --role acrpull ...').")
 
+    with self.argument_context('containerapp job logs show') as c:
+        c.argument('follow', help="Print logs in real time if present.", arg_type=get_three_state_flag())
+        c.argument('tail', help="The number of past logs to print (0-300)", type=int, default=20)
+        c.argument('container', help="The name of the container")
+        c.argument('output_format', options_list=["--format"], help="Log output format", arg_type=get_enum_type(["json", "text"]), default="json")
+        c.argument('replica', help="The name of the replica. List replicas with 'az containerapp job replica list'. A replica may not exist if the job pod has been cleaned up.")
+        c.argument('execution', help="The name of the container app execution. Defaults to the latest execution.")
+        c.argument('name', name_type, id_part=None, help="The name of the Containerapp job.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+
+    with self.argument_context('containerapp job replica') as c:
+        c.argument('replica', help="The name of the replica. ")
+        c.argument('execution', help="The name of the container app execution. Defaults to the latest execution.")
+        c.argument('name', name_type, id_part=None, help="The name of the Containerapp.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+
     with self.argument_context('containerapp env dotnet-component') as c:
         c.argument('dotnet_component_name', options_list=['--name', '-n'], help="The DotNet component name.")
         c.argument('environment_name', options_list=['--environment'], help="The environment name.")
@@ -440,3 +456,12 @@ def load_arguments(self, _):
         c.argument('logger_name', help="The logger name.")
         c.argument('logger_level', arg_type=get_enum_type(["off", "error", "info", "debug", "trace", "warn"]), help="Set the log level for the specific logger name.")
         c.argument('all', help="The flag to indicate all logger settings.", action="store_true")
+
+    with self.argument_context('containerapp diagnose dns resolve') as c:
+        c.argument('name', options_list=['--name', '-n'], help="The managed environment name.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+        c.argument('domain', help="The domain to check DNS resolution for. Defaults to azure.com")
+
+    with self.argument_context('containerapp diagnose healthcheck show') as c:
+        c.argument('name', options_list=['--name', '-n'], help="The managed environment name.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)

src/containerapp/azext_containerapp/commands.py

@@ -256,3 +256,9 @@ def load_command_table(self, args):
         g.custom_command('set', 'create_or_update_java_logger', supports_no_wait=True)
         g.custom_command('delete', 'delete_java_logger', supports_no_wait=True)
         g.custom_show_command('show', 'show_java_logger')
+
+    with self.command_group('containerapp diagnose dns', is_preview=True) as g:
+        g.custom_command('resolve', 'diagnose_dns')
+
+    with self.command_group('containerapp diagnose healthcheck', is_preview=True) as g:
+        g.custom_show_command('show', 'healthcheck_show')

src/containerapp/azext_containerapp/custom.py

@@ -3224,3 +3224,73 @@ def set_registry_job(cmd, name, resource_group_name, server, username=None, pass
     containerapp_job_registry_set_decorator.construct_payload()
     r = containerapp_job_registry_set_decorator.set()
     return r
+
+
+def diagnose_dns(cmd, resource_group_name, name, domain=None):
+    if not domain:
+        domain = "azure.com"
+
+    sub = get_subscription_id(cmd.cli_ctx)
+
+    token_response = ManagedEnvironmentPreviewClient.get_auth_token(cmd, resource_group_name, name)
+    token = token_response["properties"]["token"]
+
+    env = ManagedEnvironmentPreviewClient.show(cmd, resource_group_name, name)
+    base_url = env["properties"]["eventStreamEndpoint"]
+    base_url = base_url[:base_url.index("/subscriptions/")]
+
+    verify = True
+
+    # TODO: Remove this
+    base_url = "https://localhost:51748"
+    verify = False
+
+    url = f"{base_url}/subscriptions/{sub}/resourceGroups/{resource_group_name}/managedEnvironments/{name}/testdnsresolution?url={domain}"
+
+
+    headers = {"Authorization": f"Bearer {token}"}
+    try:
+        resp = requests.post(url,
+                            timeout=None,
+                            headers=headers,
+                            verify=verify)
+
+    except requests.exceptions.RequestException as e:
+        raise ValidationError(f"Request failed: {e}")
+
+    display_healthcheck_results(resp.content)
+
+def healthcheck_show(cmd, resource_group_name, name):
+    sub = get_subscription_id(cmd.cli_ctx)
+
+    token_response = ManagedEnvironmentPreviewClient.get_auth_token(cmd, resource_group_name, name)
+    token = token_response["properties"]["token"]
+
+    env = ManagedEnvironmentPreviewClient.show(cmd, resource_group_name, name)
+    base_url = env["properties"]["eventStreamEndpoint"]
+    base_url = base_url[:base_url.index("/subscriptions/")]
+
+    verify = True
+
+    # TODO: Remove this
+    base_url = "https://localhost:51748"
+    verify = False
+
+    url = f"{base_url}/subscriptions/{sub}/resourceGroups/{resource_group_name}/managedEnvironments/{name}/healthcheck"
+
+
+    headers = {"Authorization": f"Bearer {token}"}
+    try:
+        resp = requests.post(url,
+                            timeout=None,
+                            headers=headers,
+                            verify=verify)
+
+    except requests.exceptions.RequestException as e:
+        raise ValidationError(f"Request failed: {e}")
+
+    display_healthcheck_results(resp.content)
+
+def display_healthcheck_results(result):
+    print(result)
+