[AKS Preview] Add backup enablement support via dataprotection extension

src/aks-preview/azext_aks_preview/_params.py

@@ -23,6 +23,21 @@
     validate_nat_gateway_idle_timeout,
     validate_nat_gateway_managed_outbound_ip_count,
 )
+
+# Import backup strategy constants from dataprotection extension
+from azure.cli.core.extension.operations import add_extension_to_path
+add_extension_to_path("dataprotection")
+from azext_dataprotection.manual._consts import (
+    CONST_AKS_BACKUP_STRATEGIES,
+    CONST_BACKUP_STRATEGY_WEEK,
+    CONST_BACKUP_STRATEGY_MONTH,
+    CONST_BACKUP_STRATEGY_IMMUTABLE,
+    CONST_BACKUP_STRATEGY_DISASTER_RECOVERY,
+    CONST_BACKUP_STRATEGY_CUSTOM,
+)
+
+backup_presets = CONST_AKS_BACKUP_STRATEGIES
+
 from azure.cli.core.commands.parameters import (
     edge_zone_type,
     file_type,
@@ -163,6 +178,7 @@
     CONST_UPGRADE_STRATEGY_ROLLING,
     CONST_UPGRADE_STRATEGY_BLUE_GREEN
 )
+from azure.cli.core.commands.validators import validate_file_or_dict
 
 from azext_aks_preview._validators import (
     validate_acr,
@@ -1740,6 +1756,13 @@ def load_arguments(self, _):
                 'by that action.'
             )
         )
+        c.argument("enable_backup", help="Enable backup for the cluster", is_preview=True, action="store_true")
+        c.argument("backup_strategy", arg_type=get_enum_type(backup_presets), help="Backup strategy for the cluster. Defaults to Week.", is_preview=True)
+        c.argument("backup_configuration_file", type=validate_file_or_dict, 
+                   options_list=['--backup-configuration-file', '-f'],
+                   help="Path to backup configuration file (JSON) or inline JSON string.", is_preview=True)
+        # In update scenario, use emtpy str as default.
+        c.argument('ssh_access', arg_type=get_enum_type(ssh_accesses), is_preview=True)
         c.argument('enable_static_egress_gateway', is_preview=True, action='store_true')
         c.argument('disable_static_egress_gateway', is_preview=True, action='store_true')
         c.argument("enable_imds_restriction", action="store_true", is_preview=True)

src/aks-preview/azext_aks_preview/custom.py

@@ -1394,6 +1394,10 @@ def aks_update(
     # IMDS restriction
     enable_imds_restriction=False,
     disable_imds_restriction=False,
+    # Backup
+    enable_backup=False,
+    backup_strategy=None,
+    backup_configuration_parameters=None,
     migrate_vmas_to_vms=False,
     enable_upstream_kubescheduler_user_configuration=False,
     disable_upstream_kubescheduler_user_configuration=False,

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -6,6 +6,7 @@
 # pylint: disable=too-many-lines
 import copy
 import datetime
+import json
 import os
 from types import SimpleNamespace
 from typing import Any, Dict, List, Optional, Tuple, TypeVar, Union
@@ -7373,9 +7374,39 @@ def update_mc_profile_preview(self) -> ManagedCluster:
         mc = self.update_upstream_kubescheduler_user_configuration(mc)
         # update ManagedSystem pools, must at end
         mc = self.update_managed_system_pools(mc)
+        # set up backup
+        mc = self.set_up_backup(mc)
 
         return mc
 
+    def set_up_backup(self, mc: ManagedCluster) -> ManagedCluster:
+
+        enable_backup = self.context.raw_param.get("enable_backup")
+        if enable_backup:
+            # Validate that dataprotection extension is installed
+            try:
+                from azure.cli.core.extension.operations import add_extension_to_path
+                add_extension_to_path("dataprotection")
+                from azext_dataprotection.manual.aks.aks_helper import dataprotection_enable_backup_helper
+            except (ImportError, ModuleNotFoundError):
+                raise CLIError(
+                    "The 'dataprotection' extension is required for AKS backup functionality.\n"
+                    "Please install it using: az extension add --name dataprotection"
+                )
+
+            backup_strategy = self.context.raw_param.get("backup_strategy")
+            backup_configuration_file = self.context.raw_param.get("backup_configuration_file")
+
+            # Build the cluster resource ID
+            cluster_resource_id = (
+                f"/subscriptions/{self.context.get_subscription_id()}"
+                f"/resourceGroups/{self.context.get_resource_group_name()}"
+                f"/providers/Microsoft.ContainerService/managedClusters/{self.context.get_name()}"
+            )
+
+            dataprotection_enable_backup_helper(self.cmd, str(cluster_resource_id), backup_strategy, backup_configuration_file)
+        return mc
+
     def check_is_postprocessing_required(self, mc: ManagedCluster) -> bool:
         """Helper function to check if postprocessing is required after sending a PUT request to create the cluster.