Containerapp job support keda msi (#7712)

src/containerapp/HISTORY.rst

@@ -9,6 +9,7 @@ upcoming
 * 'az containerapp job replica list': Support list replicas of a job execution
 * 'az containerapp env update': Fix logs configuration about removing destination with `--logs-destination none`
 * 'az containerapp auth update': Fix AuthConfigSecretRefNotFound when setting secret
+* 'az containerapp job create/update': Support --scale-rule-identity for scale rule to authenticate to azure resource scaler
 
 0.3.52
 ++++++

src/containerapp/azext_containerapp/_help.py

@@ -1220,9 +1220,28 @@
               --scale-rule-type azure-queue \\
               --scale-rule-metadata "accountName=mystorageaccountname" \\
                                     "cloud=AzurePublicCloud" \\
-                                    "queueLength": "5" "queueName": "foo" \\
+                                    "queueLength=5" "queueName=foo" \\
               --scale-rule-auth "connection=my-connection-string-secret-name" \\
               --image imageName
+    - name: Create container app job with Trigger Type as Event using identity to authenticate
+      text: |
+          az containerapp job create -n MyContainerappsjob -g MyResourceGroup \\
+              --environment MyContainerappEnv
+              --trigger-type Event \\
+              --replica-timeout 5 \\
+              --replica-retry-limit 2 \\
+              --replica-completion-count 1 \\
+              --parallelism 1 \\
+              --polling-interval 30 \\
+              --min-executions 0 \\
+              --max-executions 1 \\
+              --scale-rule-name azure-queue \\
+              --scale-rule-type azure-queue \\
+              --scale-rule-metadata "accountName=mystorageaccountname" \\
+                                    "cloud=AzurePublicCloud" \\
+                                    "queueLength=5" "queueName=foo" \\
+              --scale-rule-identity myUserIdentityResourceId \\
+              --image imageName
     - name: Create a container apps job hosted on a Connected Environment.
       text: |
           az containerapp job create -n MyContainerappsjob -g MyResourceGroup \\

src/containerapp/azext_containerapp/_params.py

@@ -305,6 +305,7 @@ def load_arguments(self, _):
         c.argument('max_executions', type=int, help="Maximum number of job executions to run per polling interval.")
         c.argument('polling_interval', type=int, help="Interval to check each event source in seconds. Defaults to 30s.")
         c.argument('scale_rule_type', options_list=['--scale-rule-type', '--srt'], help="The type of the scale rule.")
+        c.argument('scale_rule_identity', options_list=['--scale-rule-identity', '--sri'], help='Resource ID of a managed identity to authenticate with Azure scaler resource(storage account/eventhub or else), or System to use a system-assigned identity.', is_preview=True)
 
     # params for preview
     with self.argument_context('containerapp') as c:

src/containerapp/azext_containerapp/containerapp_job_decorator.py

@@ -8,14 +8,15 @@
 from urllib.parse import urlparse
 
 from azure.cli.core.azclierror import (
-    RequiredArgumentMissingError, ValidationError)
+    RequiredArgumentMissingError, ValidationError, InvalidArgumentValueError)
 from azure.cli.command_modules.containerapp.containerapp_job_decorator import ContainerAppJobCreateDecorator, \
     ContainerAppJobDecorator
 from azure.cli.command_modules.containerapp._utils import safe_get, _convert_object_from_snake_to_camel_case, \
     _object_to_dict, _remove_additional_attributes, _remove_readonly_attributes, clean_null_values, \
     _populate_secret_values, _add_or_update_tags, ensure_workload_profile_supported, _add_or_update_env_vars, \
     parse_env_var_flags, _remove_env_vars, _get_acr_cred, store_as_secret_and_return_secret_ref, \
-    parse_metadata_flags, parse_auth_flags, safe_set
+    parse_metadata_flags, parse_auth_flags, safe_set, _ensure_identity_resource_id
+from azure.cli.core.commands.client_factory import get_subscription_id
 from azure.cli.core.commands import AzCliCommand
 from azure.core.exceptions import DeserializationError, ResourceNotFoundError
 
@@ -328,7 +329,7 @@ def set_up_trigger_configurations(self):
                     if self.get_argument_parallelism():
                         event_trigger_config_def["parallelism"] = self.get_argument_parallelism()
                     # Scale
-                    if "scale" in event_trigger_config_def["scale"]:
+                    if "scale" not in event_trigger_config_def:
                         event_trigger_config_def["scale"] = {}
                     if self.get_argument_min_executions() is not None:
                         event_trigger_config_def["scale"]["minExecutions"] = self.get_argument_min_executions()
@@ -502,12 +503,24 @@ class ContainerAppJobPreviewCreateDecorator(ContainerAppJobCreateDecorator):
     def construct_payload(self):
         super().construct_payload()
         self.set_up_extended_location()
+        if self.get_argument_scale_rule_identity():
+            scaleRules = safe_get(self.containerappjob_def, "properties", "configuration", "eventTriggerConfig", "scale", "rules", default=[])
+            if scaleRules and len(scaleRules) > 0:
+                identity = self.get_argument_scale_rule_identity().lower()
+                if identity != "system":
+                    subscription_id = get_subscription_id(self.cmd.cli_ctx)
+                    identity = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), identity)
+                self.containerappjob_def["properties"]["configuration"]["eventTriggerConfig"]["scale"]["rules"][0]["identity"] = identity
 
     def validate_arguments(self):
         super().validate_arguments()
         if self.get_argument_yaml() is None:
             if self.get_argument_trigger_type() is None:
                 raise RequiredArgumentMissingError('Usage error: --trigger-type is required')
+        if self.get_argument_scale_rule_type() and self.get_argument_scale_rule_identity():
+            scale_rule_type = self.get_argument_scale_rule_type().lower()
+            if scale_rule_type == "http" or scale_rule_type == "tcp":
+                raise InvalidArgumentValueError("--scale-rule-identity cannot be set when --scale-rule-type is 'http' or 'tcp'")
 
     def set_up_extended_location(self):
         if self.get_argument_environment_type() == CONNECTED_ENVIRONMENT_TYPE:
@@ -552,6 +565,9 @@ def get_environment_client(self):
     def get_argument_environment_type(self):
         return self.get_param("environment_type")
 
+    def get_argument_scale_rule_identity(self):
+        return self.get_param("scale_rule_identity")
+
     def set_argument_managed_env(self, managed_env):
         self.set_param("managed_env", managed_env)
 
@@ -563,3 +579,35 @@ class ContainerAppJobPreviewUpdateDecorator(ContainerAppJobUpdateDecorator):
     # pylint: disable=useless-super-delegation
     def construct_payload(self):
         super().construct_payload()
+
+    def validate_arguments(self):
+        super().validate_arguments()
+        if self.get_argument_scale_rule_type() and self.get_argument_scale_rule_identity():
+            scale_rule_type = self.get_argument_scale_rule_type().lower()
+            if scale_rule_type == "http" or scale_rule_type == "tcp":
+                raise InvalidArgumentValueError("--scale-rule-identity cannot be set when --scale-rule-type is 'http' or 'tcp'")
+
+    def set_up_trigger_configurations(self):
+        super().set_up_trigger_configurations()
+        identity = self.get_argument_scale_rule_identity()
+        if identity:
+            trigger_type = safe_get(self.containerappjob_def, "properties", "configuration", "triggerType")
+            if trigger_type == "Event":
+                existing_rules = safe_get(self.containerappjob_def, "properties", "configuration", "eventTriggerConfig", "scale", "rules", default=[])
+                if existing_rules and len(existing_rules) > 0:
+                    identity = self.get_argument_scale_rule_identity().lower()
+                    if identity != "system":
+                        subscription_id = get_subscription_id(self.cmd.cli_ctx)
+                        identity = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), identity)
+                    for rule in existing_rules:
+                        if rule["name"] == self.get_argument_scale_rule_name():
+                            rule["identity"] = identity
+                            break
+                    safe_set(self.new_containerappjob, "properties", "configuration", "eventTriggerConfig", "scale", "rules", value=existing_rules)
+
+    def should_update_trigger_configurations(self):
+        return super().should_update_trigger_configurations() \
+            or self.get_argument_scale_rule_identity()
+
+    def get_argument_scale_rule_identity(self):
+        return self.get_param("scale_rule_identity")

src/containerapp/azext_containerapp/custom.py

@@ -917,6 +917,7 @@ def create_containerappsjob(cmd,
                             scale_rule_name=None,
                             scale_rule_type=None,
                             scale_rule_auth=None,
+                            scale_rule_identity=None,
                             polling_interval=30,
                             min_executions=0,
                             max_executions=10,
@@ -969,6 +970,7 @@ def update_containerappsjob(cmd,
                             scale_rule_name=None,
                             scale_rule_type=None,
                             scale_rule_auth=None,
+                            scale_rule_identity=None,
                             polling_interval=None,
                             min_executions=None,
                             max_executions=None,