Add k8s-extension troubleshoot phase 1: Infrastructure setup.

src/k8s-extension/azext_k8s_extension/__init__.py

@@ -5,9 +5,11 @@
 
 from azure.cli.core import AzCommandsLoader
 from . import consts
+from typing import Union
 
 from ._help import helps  # pylint: disable=unused-import
 
+from knack.commands import CLICommand
 
 class K8sExtensionCommandsLoader(AzCommandsLoader):
 
@@ -20,7 +22,7 @@ def __init__(self, cli_ctx=None):
         super().__init__(cli_ctx=cli_ctx,
                          custom_command_type=k8s_extension_custom)
 
-    def load_command_table(self, args):
+    def load_command_table(self, args: Union[list[str], None]) -> dict[str, CLICommand]:
         from .commands import load_command_table
         from azure.cli.core.aaz import load_aaz_command_table
         try:
@@ -34,9 +36,10 @@ def load_command_table(self, args):
                 args=args
             )
         load_command_table(self, args)
-        return self.command_table
+        command_table: dict[str, CLICommand] = self.command_table
+        return command_table
 
-    def load_arguments(self, command):
+    def load_arguments(self, command: CLICommand):
         from ._params import load_arguments
         load_arguments(self, command)
 

src/k8s-extension/azext_k8s_extension/_help.py

@@ -94,6 +94,18 @@
 --config-protected-file=protected-settings-file
 """
 
+helps[f'{consts.EXTENSION_NAME} troubleshoot'] = f"""
+    type: command
+    short-summary: Perform diagnostic checks on a Kubernetes Extension.
+    long-summary: This command is used to troubleshoot a Kubernetes Extension. It \
+collects logs and other information that can be used to diagnose issues with the extension.
+    examples:
+      - name: Troubleshoot a Kubernetes Extension
+        text: |-
+          az {consts.EXTENSION_NAME} troubleshoot --name extension-name \
+--namespace-list "namespace1,namespace2"
+"""
+
 helps[f'{consts.EXTENSION_NAME} extension-types'] = """
     type: group
     short-summary: Commands to discover Kubernetes Extension Types.

src/k8s-extension/azext_k8s_extension/_params.py

@@ -15,8 +15,9 @@
     AddConfigurationProtectedSettings,
 )
 
+from knack.commands import CLICommand
 
-def load_arguments(self, _):
+def load_arguments(self, _: CLICommand) -> None:
     with self.argument_context(consts.EXTENSION_NAME) as c:
         c.argument('location',
                    validator=get_default_location_from_resource_group)
@@ -131,3 +132,20 @@ def load_arguments(self, _):
         c.argument('show_latest',
                    arg_type=get_three_state_flag(),
                    help='Filter results by only the latest version. For example, if this flag is used the latest version of the extensionType will be shown.')
+
+    with self.argument_context(f"{consts.EXTENSION_NAME} troubleshoot") as c:
+        c.argument('name',
+                   options_list=['--name', '-n'],
+                   help='Name of the Kubernetes extension')
+        c.argument('namespace_list',
+                   options_list=['--namespace-list'],
+                   help='Comma-separated list of namespaces to troubleshoot')
+        c.argument('kube_config',
+                   options_list=['--kube-config'],
+                   help='Path to the kube config file. If not specified, the default kube config file will be used.')
+        c.argument('kube_context',
+                   options_list=['--kube-context'],
+                   help='Kubeconfig context from current machine. If not specified, the current context from kube config file will be used.')
+        c.argument('skip_ssl_verification',
+                   action="store_true",
+                   help='Skip SSL verification for any cluster connection.')

src/k8s-extension/azext_k8s_extension/commands.py

@@ -23,6 +23,7 @@ def load_command_table(self, _):
         g.custom_command('list', 'list_k8s_extension', table_transformer=k8s_extension_list_table_format)
         g.custom_show_command('show', 'show_k8s_extension', table_transformer=k8s_extension_show_table_format)
         g.custom_command('update', 'update_k8s_extension', supports_no_wait=True)
+        g.custom_command('troubleshoot', 'troubleshoot_extension', is_preview=True)
 
     # Subgroup - k8s-extension extension-types
     k8s_cluster_extension_type_sdk = CliCommandType(

src/k8s-extension/azext_k8s_extension/consts.py

@@ -26,3 +26,27 @@
 HYBRIDCONTAINERSERVICE_API_VERSION = "2022-05-01-preview"
 
 EXTENSION_TYPE_API_VERSION = "2023-05-01-preview"
+
+# Fault type constants for error categorization.
+# Used to classify different types of faults encountered during diagnostics.
+LOAD_KUBECONFIG_FAULT_TYPE = "kubeconfig-load-error"    # Error loading kubeconfig file.
+
+# Warning messages for diagnostic failures.
+KUBECONFIG_LOAD_FAILED_WARNING = """Unable to load the kubeconfig file.
+Please check
+https://learn.microsoft.com/en-us/azure/azure-arc/kubernetes/diagnose-connection-issues#is-kubeconfig-pointing-to-the-right-cluster"""
+
+EXTRACT_HELMEXE_FAULT_TYPE = "helm-client-extract-error"    # Error extracting Helm client executable.
+
+HELM_VERSION = "v3.12.2"
+
+DOWNLOAD_AND_INSTALL_KUBECTL_FAULT_TYPE = "Failed to download and install kubectl"  # Error downloading/installing kubectl.
+
+KUBEAPI_CONNECTIVITY_FAILED_WARNING = """Unable to verify connectivity to the Kubernetes cluster.
+Please check https://learn.microsoft.com/en-us/azure/azure-arc/kubernetes/diagnose-connection-issues"""
+
+KUBERNETES_CONNECTIVITY_FAULT_TYPE = "kubernetes-cluster-connection-error"  # Error connecting to Kubernetes cluster.
+
+# Diagnostic log file path constant.
+# Used to specify the name of the file where extension diagnostic logs are stored.
+ARC_EXT_DIAGNOSTIC_LOGS = "arc_ext_diagnostic_logs"