fix: apply resource conditions to DeploymentPreference resources

samtranslator/model/preferences/deployment_preference.py

@@ -25,7 +25,16 @@
 """
 DeploymentPreferenceTuple = namedtuple(
     "DeploymentPreferenceTuple",
-    ["deployment_type", "pre_traffic_hook", "post_traffic_hook", "alarms", "enabled", "role", "trigger_configurations"],
+    [
+        "deployment_type",
+        "pre_traffic_hook",
+        "post_traffic_hook",
+        "alarms",
+        "enabled",
+        "role",
+        "trigger_configurations",
+        "condition",
+    ],
 )
 
 
@@ -37,17 +46,18 @@ class DeploymentPreference(DeploymentPreferenceTuple):
     """
 
     @classmethod
-    def from_dict(cls, logical_id, deployment_preference_dict):
+    def from_dict(cls, logical_id, deployment_preference_dict, condition=None):
         """
         :param logical_id: the logical_id of the resource that owns this deployment preference
         :param deployment_preference_dict: the dict object taken from the SAM template
+        :param condition: condition on this deployment preference
         :return:
         """
         enabled = deployment_preference_dict.get("Enabled", True)
         enabled = False if enabled in ["false", "False"] else enabled
 
         if not enabled:
-            return DeploymentPreference(None, None, None, None, False, None, None)
+            return DeploymentPreference(None, None, None, None, False, None, None, None)
 
         if "Type" not in deployment_preference_dict:
             raise InvalidResourceException(logical_id, "'DeploymentPreference' is missing required Property 'Type'")
@@ -64,6 +74,15 @@ def from_dict(cls, logical_id, deployment_preference_dict):
         alarms = deployment_preference_dict.get("Alarms", None)
         role = deployment_preference_dict.get("Role", None)
         trigger_configurations = deployment_preference_dict.get("TriggerConfigurations", None)
+        passthrough_condition = deployment_preference_dict.get("PassthroughCondition", False)
+
         return DeploymentPreference(
-            deployment_type, pre_traffic_hook, post_traffic_hook, alarms, enabled, role, trigger_configurations
+            deployment_type,
+            pre_traffic_hook,
+            post_traffic_hook,
+            alarms,
+            enabled,
+            role,
+            trigger_configurations,
+            condition if passthrough_condition else None,
         )

samtranslator/model/preferences/deployment_preference_collection.py

@@ -9,6 +9,9 @@
     is_intrinsic_if,
     is_intrinsic_no_value,
     validate_intrinsic_if_items,
+    make_combined_condition,
+    ref,
+    fnGetAtt,
 )
 from samtranslator.model.update_policy import UpdatePolicy
 from samtranslator.translator.arn_generator import ArnGenerator
@@ -27,6 +30,7 @@
     "Linear10PercentEvery10Minutes",
     "AllAtOnce",
 ]
+CODE_DEPLOY_CONDITION_NAME = "ServerlessCodeDeployCondition"
 
 
 class DeploymentPreferenceCollection(object):
@@ -41,20 +45,19 @@ class DeploymentPreferenceCollection(object):
 
     def __init__(self):
         """
-        This collection stores an intenral dict of the deployment preferences for each function's
-            deployment preference in the SAM Template.
+        This collection stores an internal dict of the deployment preferences for each function's
+        deployment preference in the SAM Template.
         """
         self._resource_preferences = {}
-        self.codedeploy_application = self._codedeploy_application()
-        self.codedeploy_iam_role = self._codedeploy_iam_role()
 
-    def add(self, logical_id, deployment_preference_dict):
+    def add(self, logical_id, deployment_preference_dict, condition=None):
         """
         Add this deployment preference to the collection
 
         :raise ValueError if an existing logical id already exists in the _resource_preferences
         :param logical_id: logical id of the resource where this deployment preference applies
         :param deployment_preference_dict: the input SAM template deployment preference mapping
+        :param condition: the condition (if it exists) on the serverless function
         """
         if logical_id in self._resource_preferences:
             raise ValueError(
@@ -63,7 +66,9 @@ def add(self, logical_id, deployment_preference_dict):
                 )
             )
 
-        self._resource_preferences[logical_id] = DeploymentPreference.from_dict(logical_id, deployment_preference_dict)
+        self._resource_preferences[logical_id] = DeploymentPreference.from_dict(
+            logical_id, deployment_preference_dict, condition
+        )
 
     def get(self, logical_id):
         """
@@ -85,18 +90,52 @@ def can_skip_service_role(self):
         """
         return all(preference.role or not preference.enabled for preference in self._resource_preferences.values())
 
+    def needs_resource_condition(self):
+        """
+        If all preferences have a condition, all code deploy resources need to be conditionally created
+        :return: True, if a condition needs to be created
+        """
+        # If there are any enabled deployment preferences without conditions, return false
+        return self._resource_preferences and not any(
+            not preference.condition and preference.enabled for preference in self._resource_preferences.values()
+        )
+
+    def get_all_deployment_conditions(self):
+        """
+        Returns a list of all conditions associated with the deployment preference resources
+        :return: List of condition names
+        """
+        conditions_set = set([preference.condition for preference in self._resource_preferences.values()])
+        if None in conditions_set:
+            # None can exist if there are disabled deployment preference(s)
+            conditions_set.remove(None)
+        return list(conditions_set)
+
+    def create_aggregate_deployment_condition(self):
+        """
+        Creates an aggregate deployment condition if necessary
+        :return: None if <2 conditions are found, otherwise a dictionary of new conditions to add to template
+        """
+        return make_combined_condition(self.get_all_deployment_conditions(), CODE_DEPLOY_CONDITION_NAME)
+
     def enabled_logical_ids(self):
         """
         :return: only the logical id's for the deployment preferences in this collection which are enabled
         """
         return [logical_id for logical_id, preference in self._resource_preferences.items() if preference.enabled]
 
-    def _codedeploy_application(self):
+    def get_codedeploy_application(self):
         codedeploy_application_resource = CodeDeployApplication(CODEDEPLOY_APPLICATION_LOGICAL_ID)
         codedeploy_application_resource.ComputePlatform = "Lambda"
+        if self.needs_resource_condition():
+            conditions = self.get_all_deployment_conditions()
+            condition_name = CODE_DEPLOY_CONDITION_NAME
+            if len(conditions) <= 1:
+                condition_name = conditions.pop()
+            codedeploy_application_resource.set_resource_attribute("Condition", condition_name)
         return codedeploy_application_resource
 
-    def _codedeploy_iam_role(self):
+    def get_codedeploy_iam_role(self):
         iam_role = IAMRole(CODE_DEPLOY_SERVICE_ROLE_LOGICAL_ID)
         iam_role.AssumeRolePolicyDocument = {
             "Version": "2012-10-17",
@@ -120,6 +159,12 @@ def _codedeploy_iam_role(self):
                 ArnGenerator.generate_aws_managed_policy_arn("service-role/AWSCodeDeployRoleForLambda")
             ]
 
+        if self.needs_resource_condition():
+            conditions = self.get_all_deployment_conditions()
+            condition_name = CODE_DEPLOY_CONDITION_NAME
+            if len(conditions) <= 1:
+                condition_name = conditions.pop()
+            iam_role.set_resource_attribute("Condition", condition_name)
         return iam_role
 
     def deployment_group(self, function_logical_id):
@@ -137,7 +182,7 @@ def deployment_group(self, function_logical_id):
         except ValueError as e:
             raise InvalidResourceException(function_logical_id, str(e))
 
-        deployment_group.ApplicationName = self.codedeploy_application.get_runtime_attr("name")
+        deployment_group.ApplicationName = ref(CODEDEPLOY_APPLICATION_LOGICAL_ID)
         deployment_group.AutoRollbackConfiguration = {
             "Enabled": True,
             "Events": ["DEPLOYMENT_FAILURE", "DEPLOYMENT_STOP_ON_ALARM", "DEPLOYMENT_STOP_ON_REQUEST"],
@@ -149,13 +194,16 @@ def deployment_group(self, function_logical_id):
 
         deployment_group.DeploymentStyle = {"DeploymentType": "BLUE_GREEN", "DeploymentOption": "WITH_TRAFFIC_CONTROL"}
 
-        deployment_group.ServiceRoleArn = self.codedeploy_iam_role.get_runtime_attr("arn")
+        deployment_group.ServiceRoleArn = fnGetAtt(CODE_DEPLOY_SERVICE_ROLE_LOGICAL_ID, "Arn")
         if deployment_preference.role:
             deployment_group.ServiceRoleArn = deployment_preference.role
 
         if deployment_preference.trigger_configurations:
             deployment_group.TriggerConfigurations = deployment_preference.trigger_configurations
 
+        if deployment_preference.condition:
+            deployment_group.set_resource_attribute("Condition", deployment_preference.condition)
+
         return deployment_group
 
     def _convert_alarms(self, preference_alarms):
@@ -240,7 +288,7 @@ def update_policy(self, function_logical_id):
         deployment_preference = self.get(function_logical_id)
 
         return UpdatePolicy(
-            self.codedeploy_application.get_runtime_attr("name"),
+            ref(CODEDEPLOY_APPLICATION_LOGICAL_ID),
             self.deployment_group(function_logical_id).get_runtime_attr("name"),
             deployment_preference.pre_traffic_hook,
             deployment_preference.post_traffic_hook,

samtranslator/model/sam_resources.py

@@ -1,5 +1,7 @@
 """ SAM macro definitions """
 import copy
+from typing import Union
+from samtranslator.intrinsics.resolver import IntrinsicsResolver
 
 import samtranslator.model.eventsources
 import samtranslator.model.eventsources.pull
@@ -139,6 +141,7 @@ def to_cloudformation(self, **kwargs):
         intrinsics_resolver = kwargs["intrinsics_resolver"]
         mappings_resolver = kwargs.get("mappings_resolver", None)
         conditions = kwargs.get("conditions", {})
+        feature_toggle = kwargs.get("feature_toggle")
 
         if self.DeadLetterQueue:
             self._validate_dlq()
@@ -185,6 +188,8 @@ def to_cloudformation(self, **kwargs):
                 lambda_alias,
                 intrinsics_resolver,
                 mappings_resolver,
+                self.get_passthrough_resource_attributes(),
+                feature_toggle,
             )
         event_invoke_policies = []
         if self.EventInvokeConfig:
@@ -827,10 +832,16 @@ def _construct_alias(self, name, function, version):
         return alias
 
     def _validate_deployment_preference_and_add_update_policy(
-        self, deployment_preference_collection, lambda_alias, intrinsics_resolver, mappings_resolver
+        self,
+        deployment_preference_collection,
+        lambda_alias,
+        intrinsics_resolver,
+        mappings_resolver,
+        passthrough_resource_attributes,
+        feature_toggle=None,
     ):
         if "Enabled" in self.DeploymentPreference:
-            # resolve intrinsics and mappings for Type
+            # resolve intrinsics and mappings for Enabled
             enabled = self.DeploymentPreference["Enabled"]
             enabled = intrinsics_resolver.resolve_parameter_refs(enabled)
             enabled = mappings_resolver.resolve_parameter_refs(enabled)
@@ -843,10 +854,28 @@ def _validate_deployment_preference_and_add_update_policy(
             preference_type = mappings_resolver.resolve_parameter_refs(preference_type)
             self.DeploymentPreference["Type"] = preference_type
 
+        if "PassthroughCondition" in self.DeploymentPreference:
+            self.DeploymentPreference["PassthroughCondition"] = self._resolve_property_to_boolean(
+                self.DeploymentPreference["PassthroughCondition"],
+                "PassthroughCondition",
+                intrinsics_resolver,
+                mappings_resolver,
+            )
+        elif feature_toggle:
+            self.DeploymentPreference["PassthroughCondition"] = feature_toggle.is_enabled(
+                "deployment_preference_condition_fix"
+            )
+        else:
+            self.DeploymentPreference["PassthroughCondition"] = False
+
         if deployment_preference_collection is None:
             raise ValueError("deployment_preference_collection required for parsing the deployment preference")
 
-        deployment_preference_collection.add(self.logical_id, self.DeploymentPreference)
+        deployment_preference_collection.add(
+            self.logical_id,
+            self.DeploymentPreference,
+            passthrough_resource_attributes.get("Condition"),
+        )
 
         if deployment_preference_collection.get(self.logical_id).enabled:
             if self.AutoPublishAlias is None:
@@ -860,6 +889,40 @@ def _validate_deployment_preference_and_add_update_policy(
                 "UpdatePolicy", deployment_preference_collection.update_policy(self.logical_id).to_dict()
             )
 
+    def _resolve_property_to_boolean(
+        self,
+        property_value: Union[bool, str, dict],
+        property_name: str,
+        intrinsics_resolver: IntrinsicsResolver,
+        mappings_resolver: IntrinsicsResolver,
+    ) -> bool:
+        """
+        Resolves intrinsics, if any, and/or converts string in a given property to boolean.
+        Raises InvalidResourceException if can't resolve intrinsic or can't resolve string to boolean
+
+        :param property_value: property value to resolve
+        :param property_name: name/key of property to resolve
+        :param intrinsics_resolver: resolves intrinsics
+        :param mappings_resolver: resolves FindInMap
+        :return bool: resolved boolean value
+        """
+        processed_property_value = intrinsics_resolver.resolve_parameter_refs(property_value)
+        processed_property_value = mappings_resolver.resolve_parameter_refs(processed_property_value)
+
+        # FIXME: We should support not only true/false, but also yes/no, on/off? See https://yaml.org/type/bool.html
+        if processed_property_value in [True, "true", "True"]:
+            return True
+        elif processed_property_value in [False, "false", "False"]:
+            return False
+        elif is_intrinsic(processed_property_value):  # couldn't resolve intrinsic
+            raise InvalidResourceException(
+                self.logical_id,
+                f"Unsupported intrinsic: the only intrinsic functions supported for "
+                f"property {property_name} are FindInMap and parameter Refs.",
+            )
+        else:
+            raise InvalidResourceException(self.logical_id, f"Invalid value for property {property_name}.")
+
     def _construct_function_url(self, lambda_function, lambda_alias):
         """
         This method is used to construct a lambda url resource

samtranslator/translator/translator.py

@@ -142,6 +142,7 @@ def translate(self, sam_template, parameter_values, feature_toggle=None, passthr
                 )
                 kwargs["redeploy_restapi_parameters"] = self.redeploy_restapi_parameters
                 kwargs["shared_api_usage_plan"] = shared_api_usage_plan
+                kwargs["feature_toggle"] = self.feature_toggle
                 translated = macro.to_cloudformation(**kwargs)
 
                 supported_resource_refs = macro.get_resource_references(translated, supported_resource_refs)
@@ -167,10 +168,14 @@ def translate(self, sam_template, parameter_values, feature_toggle=None, passthr
                 document_errors.append(e)
 
         if deployment_preference_collection.any_enabled():
-            template["Resources"].update(deployment_preference_collection.codedeploy_application.to_dict())
+            template["Resources"].update(deployment_preference_collection.get_codedeploy_application().to_dict())
+            if deployment_preference_collection.needs_resource_condition():
+                new_conditions = deployment_preference_collection.create_aggregate_deployment_condition()
+                if new_conditions:
+                    template.get("Conditions").update(new_conditions)
 
             if not deployment_preference_collection.can_skip_service_role():
-                template["Resources"].update(deployment_preference_collection.codedeploy_iam_role.to_dict())
+                template["Resources"].update(deployment_preference_collection.get_codedeploy_iam_role().to_dict())
 
             for logical_id in deployment_preference_collection.enabled_logical_ids():
                 try: