Skip to content

[dataprotection] Add controller and initContainer resource settings for AKS Backup Extension #9648

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
priyansh17 wants to merge 1 commit into
base: main
Choose a base branch
from
+237 −0
Open

[dataprotection] Add controller and initContainer resource settings for AKS Backup Extension #9648

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions src/k8s-extension/azext_k8s_extension/partner_extensions/DataProtectionKubernetes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@ def __init__(self):
- Backup storage location (required)
- Resource Requests (optional)
- Resource Limits (optional)
- Controller Resource Requests (optional)
- Controller Resource Limits (optional)
- Init Container (veleroPluginForMicrosoftAzure) Resource Requests (optional)
- Init Container (veleroPluginForMicrosoftAzure) Resource Limits (optional)
- Disable Informer Cache (optional)
"""
self.TENANT_ID = "credentials.tenantId"
Expand All @@ -33,6 +37,14 @@ def __init__(self):
self.RESOURCE_REQUEST_MEMORY = "resources.requests.memory"
self.RESOURCE_LIMIT_CPU = "resources.limits.cpu"
self.RESOURCE_LIMIT_MEMORY = "resources.limits.memory"
self.CONTROLLER_RESOURCE_REQUEST_CPU = "controller.resources.requests.cpu"
self.CONTROLLER_RESOURCE_REQUEST_MEMORY = "controller.resources.requests.memory"
self.CONTROLLER_RESOURCE_LIMIT_CPU = "controller.resources.limits.cpu"
self.CONTROLLER_RESOURCE_LIMIT_MEMORY = "controller.resources.limits.memory"
self.PLUGIN_RESOURCE_REQUEST_CPU = "initContainers.veleroPluginForMicrosoftAzure.resources.requests.cpu"
self.PLUGIN_RESOURCE_REQUEST_MEMORY = "initContainers.veleroPluginForMicrosoftAzure.resources.requests.memory"
self.PLUGIN_RESOURCE_LIMIT_CPU = "initContainers.veleroPluginForMicrosoftAzure.resources.limits.cpu"
self.PLUGIN_RESOURCE_LIMIT_MEMORY = "initContainers.veleroPluginForMicrosoftAzure.resources.limits.memory"
self.BACKUP_STORAGE_ACCOUNT_USE_AAD = "configuration.backupStorageLocation.config.useAAD"
self.BACKUP_STORAGE_ACCOUNT_STORAGE_ACCOUNT_URI = "configuration.backupStorageLocation.config.storageAccountURI"
self.DISABLE_INFORMER_CACHE = "configuration.disableInformerCache"
Expand All @@ -45,6 +57,14 @@ def __init__(self):
self.memory_request = "memoryRequest"
self.cpu_limit = "cpuLimit"
self.memory_limit = "memoryLimit"
self.controller_cpu_request = "controllerCpuRequest"
self.controller_memory_request = "controllerMemoryRequest"
self.controller_cpu_limit = "controllerCpuLimit"
self.controller_memory_limit = "controllerMemoryLimit"
self.plugin_cpu_request = "pluginCpuRequest"
self.plugin_memory_request = "pluginMemoryRequest"
self.plugin_cpu_limit = "pluginCpuLimit"
self.plugin_memory_limit = "pluginMemoryLimit"
self.use_aad = "useAAD"
self.storage_account_uri = "storageAccountURI"
self.disable_informer_cache = "disableInformerCache"
Expand All @@ -58,6 +78,14 @@ def __init__(self):
self.memory_request.lower(): self.RESOURCE_REQUEST_MEMORY,
self.cpu_limit.lower(): self.RESOURCE_LIMIT_CPU,
self.memory_limit.lower(): self.RESOURCE_LIMIT_MEMORY,
self.controller_cpu_request.lower(): self.CONTROLLER_RESOURCE_REQUEST_CPU,
self.controller_memory_request.lower(): self.CONTROLLER_RESOURCE_REQUEST_MEMORY,
self.controller_cpu_limit.lower(): self.CONTROLLER_RESOURCE_LIMIT_CPU,
self.controller_memory_limit.lower(): self.CONTROLLER_RESOURCE_LIMIT_MEMORY,
self.plugin_cpu_request.lower(): self.PLUGIN_RESOURCE_REQUEST_CPU,
self.plugin_memory_request.lower(): self.PLUGIN_RESOURCE_REQUEST_MEMORY,
self.plugin_cpu_limit.lower(): self.PLUGIN_RESOURCE_LIMIT_CPU,
self.plugin_memory_limit.lower(): self.PLUGIN_RESOURCE_LIMIT_MEMORY,
self.use_aad.lower(): self.BACKUP_STORAGE_ACCOUNT_USE_AAD,
self.storage_account_uri.lower(): self.BACKUP_STORAGE_ACCOUNT_STORAGE_ACCOUNT_URI,
self.disable_informer_cache.lower(): self.DISABLE_INFORMER_CACHE
Expand Down Expand Up @@ -199,10 +227,17 @@ def __validate_and_map_config(self, configuration_settings, validate_bsl=True):
if key.lower() not in input_configuration_keys:
raise RequiredArgumentMissingError(f"Missing required configuration setting: {key}")

recognized_full_paths = set(self.configuration_mapping.values())
recognized_full_paths_lower = {p.lower(): p for p in recognized_full_paths}

for key in input_configuration_settings:
_key = key.lower()
if _key in self.configuration_mapping:
configuration_settings[self.configuration_mapping[_key]] = configuration_settings.pop(key).strip()
elif _key in recognized_full_paths_lower:
# User provided the full configuration path directly - normalize to canonical form
canonical = recognized_full_paths_lower[_key]
configuration_settings[canonical] = configuration_settings.pop(key).strip()
Comment on lines 233 to +240
Copy link

Copilot AI Mar 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When both a short-name key and its corresponding canonical full-path key are provided (e.g., cpuLimit and resources.limits.cpu), the current loop can overwrite one with the other based on dict insertion order, and may even pop the value that was just normalized. Please make precedence explicit (for example: prefer the explicit full-path key, or detect duplicates and raise/ignore deterministically) to avoid surprising configuration results.

Copilot uses AI. Check for mistakes.
else:
configuration_settings.pop(key)
logger.warning(f"Ignoring unrecognized configuration setting: {key}")
Expand Down
202 changes: 202 additions & 0 deletions src/k8s-extension/azext_k8s_extension/tests/latest/test_data_protection_kubernetes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,202 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

import unittest
from unittest.mock import patch
Copy link

Copilot AI Mar 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

patch is imported but never used in this test module. This will be flagged by linting and should be removed (or used if intended).

Suggested change
from unittest.mock import patch

Copilot uses AI. Check for mistakes.

from azure.cli.core.azclierror import RequiredArgumentMissingError

from azext_k8s_extension.partner_extensions.DataProtectionKubernetes import DataProtectionKubernetes


class TestDataProtectionKubernetesConfigMapping(unittest.TestCase):
def setUp(self):
self.ext = DataProtectionKubernetes()
# Access the private method via name mangling
self._validate = self.ext._DataProtectionKubernetes__validate_and_map_config

def _make_bsl_settings(self):
return {
"blobContainer": "mycontainer",
"storageAccount": "myaccount",
"storageAccountResourceGroup": "myrg",
"storageAccountSubscriptionId": "mysub",
}

# ------------------------------------------------------------------
# Short-name mapping tests
# ------------------------------------------------------------------

def test_short_names_mapped_to_full_paths(self):
config = self._make_bsl_settings()
self._validate(config)
self.assertIn("configuration.backupStorageLocation.bucket", config)
self.assertIn("configuration.backupStorageLocation.config.storageAccount", config)
self.assertIn("configuration.backupStorageLocation.config.resourceGroup", config)
self.assertIn("configuration.backupStorageLocation.config.subscriptionId", config)

def test_cpu_and_memory_limit_short_names(self):
config = {**self._make_bsl_settings(), "cpuLimit": "500m", "memoryLimit": "256Mi"}
self._validate(config)
self.assertEqual(config.get("resources.limits.cpu"), "500m")
self.assertEqual(config.get("resources.limits.memory"), "256Mi")

def test_cpu_and_memory_request_short_names(self):
config = {**self._make_bsl_settings(), "cpuRequest": "100m", "memoryRequest": "128Mi"}
self._validate(config)
self.assertEqual(config.get("resources.requests.cpu"), "100m")
self.assertEqual(config.get("resources.requests.memory"), "128Mi")

# ------------------------------------------------------------------
# Controller resource settings
# ------------------------------------------------------------------

def test_controller_cpu_limit_short_name(self):
config = {**self._make_bsl_settings(), "controllerCpuLimit": "1"}
self._validate(config)
self.assertEqual(config.get("controller.resources.limits.cpu"), "1")

def test_controller_memory_limit_short_name(self):
config = {**self._make_bsl_settings(), "controllerMemoryLimit": "512Mi"}
self._validate(config)
self.assertEqual(config.get("controller.resources.limits.memory"), "512Mi")

def test_controller_cpu_request_short_name(self):
config = {**self._make_bsl_settings(), "controllerCpuRequest": "200m"}
self._validate(config)
self.assertEqual(config.get("controller.resources.requests.cpu"), "200m")

def test_controller_memory_request_short_name(self):
config = {**self._make_bsl_settings(), "controllerMemoryRequest": "256Mi"}
self._validate(config)
self.assertEqual(config.get("controller.resources.requests.memory"), "256Mi")

# ------------------------------------------------------------------
# initContainers.veleroPluginForMicrosoftAzure resource settings
# ------------------------------------------------------------------

def test_plugin_cpu_limit_short_name(self):
config = {**self._make_bsl_settings(), "pluginCpuLimit": "1"}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.limits.cpu"), "1"
)

def test_plugin_memory_limit_short_name(self):
config = {**self._make_bsl_settings(), "pluginMemoryLimit": "512Mi"}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.limits.memory"), "512Mi"
)

def test_plugin_cpu_request_short_name(self):
config = {**self._make_bsl_settings(), "pluginCpuRequest": "100m"}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.requests.cpu"), "100m"
)

def test_plugin_memory_request_short_name(self):
config = {**self._make_bsl_settings(), "pluginMemoryRequest": "128Mi"}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.requests.memory"), "128Mi"
)

# ------------------------------------------------------------------
# Full-path pass-through tests
# ------------------------------------------------------------------

def test_full_path_memory_limit_passes_through(self):
config = {**self._make_bsl_settings(), "resources.limits.memory": "512Mi"}
self._validate(config)
self.assertEqual(config.get("resources.limits.memory"), "512Mi")

def test_full_path_cpu_limit_passes_through(self):
config = {**self._make_bsl_settings(), "resources.limits.cpu": "500m"}
self._validate(config)
self.assertEqual(config.get("resources.limits.cpu"), "500m")

def test_full_path_controller_cpu_limit_passes_through(self):
config = {**self._make_bsl_settings(), "controller.resources.limits.cpu": "1"}
self._validate(config)
self.assertEqual(config.get("controller.resources.limits.cpu"), "1")

def test_full_path_controller_memory_limit_passes_through(self):
config = {**self._make_bsl_settings(), "controller.resources.limits.memory": "512Mi"}
self._validate(config)
self.assertEqual(config.get("controller.resources.limits.memory"), "512Mi")

def test_full_path_plugin_cpu_limit_passes_through(self):
config = {
**self._make_bsl_settings(),
"initContainers.veleroPluginForMicrosoftAzure.resources.limits.cpu": "1",
}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.limits.cpu"), "1"
)

def test_full_path_plugin_memory_limit_passes_through(self):
config = {
**self._make_bsl_settings(),
"initContainers.veleroPluginForMicrosoftAzure.resources.limits.memory": "512Mi",
}
self._validate(config)
self.assertEqual(
config.get("initContainers.veleroPluginForMicrosoftAzure.resources.limits.memory"), "512Mi"
)

def test_full_path_strips_whitespace(self):
config = {**self._make_bsl_settings(), "resources.limits.memory": " 256Mi "}
self._validate(config)
self.assertEqual(config.get("resources.limits.memory"), "256Mi")

def test_full_path_case_insensitive(self):
config = {**self._make_bsl_settings(), "Resources.Limits.Memory": "512Mi"}
self._validate(config)
self.assertEqual(config.get("resources.limits.memory"), "512Mi")

# ------------------------------------------------------------------
# Unrecognized keys are ignored
# ------------------------------------------------------------------

def test_unrecognized_key_is_ignored(self):
config = {**self._make_bsl_settings(), "unknownSetting": "value"}
self._validate(config)
self.assertNotIn("unknownSetting", config)

# ------------------------------------------------------------------
# BSL validation
# ------------------------------------------------------------------

def test_missing_bsl_key_raises_error(self):
config = {
"storageAccount": "myaccount",
"storageAccountResourceGroup": "myrg",
"storageAccountSubscriptionId": "mysub",
# missing blobContainer
}
with self.assertRaises(RequiredArgumentMissingError):
self._validate(config)

def test_bsl_validation_skipped_when_disabled(self):
# Should not raise even with missing BSL settings
config = {"controllerCpuLimit": "1"}
self._validate(config, validate_bsl=False)
self.assertEqual(config.get("controller.resources.limits.cpu"), "1")

# ------------------------------------------------------------------
# Case-insensitive short name matching
# ------------------------------------------------------------------

def test_short_name_case_insensitive(self):
config = {**self._make_bsl_settings(), "CONTROLLERCPULIMIT": "500m"}
self._validate(config)
self.assertEqual(config.get("controller.resources.limits.cpu"), "500m")


if __name__ == "__main__":
unittest.main()
Loading