feat: Add AutoPublishAlias to StateMachine for managed traffic shifting

Author: GavinZZ

.cfnlintrc.yaml

@@ -118,6 +118,12 @@ ignore_templates:
   - tests/translator/output/**/state_machine_with_event_schedule_state.json
   - tests/translator/output/**/state_machine_with_schedule.json
   - tests/translator/output/**/state_machine_with_schedule_dlq_retry_policy.json
+  - tests/translator/output/**/state_machine_with_auto_publish_alias.json
+  - tests/translator/output/**/state_machine_with_deployment_preference_all_at_once.json
+  - tests/translator/output/**/state_machine_with_deployment_preference_canary.json
+  - tests/translator/output/**/state_machine_with_deployment_preference_linear.json
+  - tests/translator/output/**/state_machine_with_deletion_policy.json
+  - tests/translator/output/**/state_machine_with_update_replace_policy.json
   - tests/translator/output/**/globals_for_function.json  # RuntimeManagementConfig
   - tests/translator/output/**/function_with_runtime_config.json  # RuntimeManagementConfig
   - tests/translator/output/**/managed_policies_minimal.json  # Intentionally has non-existent managed policy name

samtranslator/internal/schema_source/aws_serverless_statemachine.py

@@ -171,6 +171,8 @@ class Properties(BaseModel):
     Tags: Optional[DictStrAny] = properties("Tags")
     Tracing: Optional[PassThroughProp] = properties("Tracing")
     Type: Optional[PassThroughProp] = properties("Type")
+    AutoPublishAlias: Optional[PassThroughProp]
+    DeploymentPreference: Optional[PassThroughProp]
 
 
 class Resource(ResourceAttributes):

samtranslator/model/__init__.py

@@ -74,6 +74,14 @@ def __init__(self, required: bool) -> None:
         super().__init__(required, any_type(), False)
 
 
+class MutatedPassThroughProperty(PassThroughProperty):
+    """
+    Mutated pass-through property.
+
+    SAM Translator may read and add/remove/modify the value before passing it to underlaying CFN resources.
+    """
+
+
 class GeneratedProperty(PropertyType):
     """
     Property of a generated CloudFormation resource.
@@ -631,6 +639,7 @@ def get_resource_by_logical_id(self, _input: str) -> Dict[str, Any]:
     "PropertyType",
     "Property",
     "PassThroughProperty",
+    "MutatedPassThroughProperty",
     "Resource",
     "ResourceMacro",
     "SamResourceMacro",

samtranslator/model/sam_resources.py

@@ -14,6 +14,7 @@
 from samtranslator.intrinsics.resolver import IntrinsicsResolver
 from samtranslator.metrics.method_decorator import cw_timer
 from samtranslator.model import (
+    MutatedPassThroughProperty,
     PassThroughProperty,
     Property,
     PropertyType,
@@ -1709,6 +1710,8 @@ class SamStateMachine(SamResourceMacro):
         "Policies": PropertyType(False, one_of(IS_STR, list_of(one_of(IS_STR, IS_DICT, IS_DICT)))),
         "Tracing": PropertyType(False, IS_DICT),
         "PermissionsBoundary": PropertyType(False, IS_STR),
+        "AutoPublishAlias": PassThroughProperty(False),
+        "DeploymentPreference": MutatedPassThroughProperty(False),
     }
 
     Definition: Optional[Dict[str, Any]]
@@ -1724,6 +1727,8 @@ class SamStateMachine(SamResourceMacro):
     Policies: Optional[List[Any]]
     Tracing: Optional[Dict[str, Any]]
     PermissionsBoundary: Optional[Intrinsicable[str]]
+    AutoPublishAlias: Optional[PassThrough]
+    DeploymentPreference: Optional[PassThrough]
 
     event_resolver = ResourceTypeResolver(
         samtranslator.model.stepfunctions.events,
@@ -1760,6 +1765,8 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
             resource_attributes=self.resource_attributes,
             passthrough_resource_attributes=self.get_passthrough_resource_attributes(),
             get_managed_policy_map=get_managed_policy_map,
+            auto_publish_alias=self.AutoPublishAlias,
+            deployment_preference=self.DeploymentPreference,
         )
 
         return state_machine_generator.to_cloudformation()

samtranslator/model/stepfunctions/__init__.py

@@ -1,5 +1,11 @@
-__all__ = ["StepFunctionsStateMachine", "StateMachineGenerator", "events"]
+__all__ = [
+    "StepFunctionsStateMachine",
+    "StepFunctionsStateMachineVersion",
+    "StepFunctionsStateMachineAlias",
+    "StateMachineGenerator",
+    "events",
+]
 
 from . import events
 from .generators import StateMachineGenerator
-from .resources import StepFunctionsStateMachine
+from .resources import StepFunctionsStateMachine, StepFunctionsStateMachineAlias, StepFunctionsStateMachineVersion

samtranslator/model/stepfunctions/generators.py

@@ -9,7 +9,11 @@
 from samtranslator.model.resource_policies import ResourcePolicies
 from samtranslator.model.role_utils import construct_role_for_resource
 from samtranslator.model.s3_utils.uri_parser import parse_s3_uri
-from samtranslator.model.stepfunctions.resources import StepFunctionsStateMachine
+from samtranslator.model.stepfunctions.resources import (
+    StepFunctionsStateMachine,
+    StepFunctionsStateMachineAlias,
+    StepFunctionsStateMachineVersion,
+)
 from samtranslator.model.tags.resource_tagging import get_tag_list
 from samtranslator.model.xray_utils import get_xray_managed_policy_name
 from samtranslator.utils.cfn_dynamic_references import is_dynamic_reference
@@ -48,6 +52,8 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: too-many-arguments
         resource_attributes=None,
         passthrough_resource_attributes=None,
         get_managed_policy_map=None,
+        auto_publish_alias=None,
+        deployment_preference=None,
     ):
         """
         Constructs an State Machine Generator class that generates a State Machine resource
@@ -73,6 +79,8 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: too-many-arguments
         :param tags: Tags to be associated with the State Machine resource
         :param resource_attributes: Resource attributes to add to the State Machine resource
         :param passthrough_resource_attributes: Attributes such as `Condition` that are added to derived resources
+        :param auto_publish_alias: Name of the state machine alias to automatically create and update
+        :deployment_preference: Settings to enable gradual state machine deployments
         """
         self.logical_id = logical_id
         self.depends_on = depends_on
@@ -100,6 +108,8 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: too-many-arguments
         )
         self.substitution_counter = 1
         self.get_managed_policy_map = get_managed_policy_map
+        self.auto_publish_alias = auto_publish_alias
+        self.deployment_preference = deployment_preference
 
     @cw_timer(prefix="Generator", name="StateMachine")
     def to_cloudformation(self):  # type: ignore[no-untyped-def]
@@ -152,6 +162,9 @@ def to_cloudformation(self):  # type: ignore[no-untyped-def]
         self.state_machine.TracingConfiguration = self.tracing
         self.state_machine.Tags = self._construct_tag_list()
 
+        managed_traffic_shifting_resources = self._generate_managed_traffic_shifting_resources()
+        resources.extend(managed_traffic_shifting_resources)
+
         event_resources = self._generate_event_resources()
         resources.extend(event_resources)
 
@@ -241,6 +254,72 @@ def _construct_tag_list(self) -> List[Dict[str, Any]]:
         sam_tag = {self._SAM_KEY: self._SAM_VALUE}
         return get_tag_list(sam_tag) + get_tag_list(self.tags)
 
+    def _construct_version(self) -> StepFunctionsStateMachineVersion:
+        """Constructs a state machine version resource that will be auto-published when the revision id of the state machine changes.
+
+        :return: Step Functions state machine version resource
+        """
+
+        # Unlike Lambda function versions, state machine versions do not need a hash suffix because
+        # they are always replaced when their corresponding state machine is updated.
+        # I.e. A SAM StateMachine resource will never have multiple version resources at the same time.
+        logical_id = f"{self.logical_id}Version"
+        attributes = self.passthrough_resource_attributes.copy()
+
+        # Both UpdateReplacePolicy and DeletionPolicy are needed to protect previous version from deletion
+        # to ensure gradual deployment works.
+        if "DeletionPolicy" not in attributes:
+            attributes["DeletionPolicy"] = "Retain"
+        if "UpdateReplacePolicy" not in attributes:
+            attributes["UpdateReplacePolicy"] = "Retain"
+
+        state_machine_version = StepFunctionsStateMachineVersion(logical_id=logical_id, attributes=attributes)
+        state_machine_version.StateMachineArn = self.state_machine.get_runtime_attr("arn")
+        state_machine_version.StateMachineRevisionId = self.state_machine.get_runtime_attr("state_machine_revision_id")
+
+        return state_machine_version
+
+    def _construct_alias(self, version: StepFunctionsStateMachineVersion) -> StepFunctionsStateMachineAlias:
+        """Constructs a state machine alias resource pointing to the given state machine version.
+        :return: Step Functions state machine alias resource
+        """
+        logical_id = f"{self.logical_id}Alias{self.auto_publish_alias}"
+        attributes = self.passthrough_resource_attributes
+
+        state_machine_alias = StepFunctionsStateMachineAlias(logical_id=logical_id, attributes=attributes)
+        state_machine_alias.Name = self.auto_publish_alias
+
+        state_machine_version_arn = version.get_runtime_attr("arn")
+
+        deployment_preference = {}
+        if self.deployment_preference:
+            deployment_preference = self.deployment_preference
+        else:
+            deployment_preference["Type"] = "ALL_AT_ONCE"
+
+        deployment_preference["StateMachineVersionArn"] = state_machine_version_arn
+        state_machine_alias.DeploymentPreference = deployment_preference
+
+        return state_machine_alias
+
+    def _generate_managed_traffic_shifting_resources(
+        self,
+    ) -> List[Any]:
+        """Generates and returns the version and alias resources associated with this state machine's managed traffic shifting.
+
+        :returns: a list containing the state machine's version and alias resources
+        :rtype: list
+        """
+        if not self.auto_publish_alias and not self.deployment_preference:
+            return []
+        if not self.auto_publish_alias and self.deployment_preference:
+            raise InvalidResourceException(
+                self.logical_id, "'DeploymentPreference' requires 'AutoPublishAlias' property to be specified."
+            )
+
+        state_machine_version = self._construct_version()
+        return [state_machine_version, self._construct_alias(state_machine_version)]
+
     def _generate_event_resources(self) -> List[Dict[str, Any]]:
         """Generates and returns the resources associated with this state machine's event sources.
 

samtranslator/model/stepfunctions/resources.py

@@ -33,4 +33,20 @@ class StepFunctionsStateMachine(Resource):
     runtime_attrs = {
         "arn": lambda self: ref(self.logical_id),
         "name": lambda self: fnGetAtt(self.logical_id, "Name"),
+        "state_machine_revision_id": lambda self: fnGetAtt(self.logical_id, "StateMachineRevisionId"),
     }
+
+
+class StepFunctionsStateMachineVersion(Resource):
+    resource_type = "AWS::StepFunctions::StateMachineVersion"
+
+    property_types = {"StateMachineArn": GeneratedProperty(), "StateMachineRevisionId": GeneratedProperty()}
+
+    runtime_attrs = {"arn": lambda self: ref(self.logical_id)}
+
+
+class StepFunctionsStateMachineAlias(Resource):
+    resource_type = "AWS::StepFunctions::StateMachineAlias"
+    property_types = {"Name": GeneratedProperty(), "DeploymentPreference": GeneratedProperty()}
+
+    runtime_attrs = {"arn": lambda self: ref(self.logical_id)}

samtranslator/schema/schema.json

@@ -242465,6 +242465,9 @@
     "samtranslator__internal__schema_source__aws_serverless_statemachine__Properties": {
       "additionalProperties": false,
       "properties": {
+        "AutoPublishAlias": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
         "Definition": {
           "markdownDescription": "The state machine definition is an object, where the format of the object matches the format of your AWS SAM template file, for example, JSON or YAML\\. State machine definitions adhere to the [Amazon States Language](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-amazon-states-language.html)\\.  \nFor an example of an inline state machine definition, see [Examples](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/#sam-resource-statemachine--examples.html#sam-resource-statemachine--examples)\\.  \nYou must provide either a `Definition` or a `DefinitionUri`\\.  \n*Type*: Map  \n*Required*: Conditional  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
           "title": "Definition",
@@ -242487,6 +242490,9 @@
           "markdownDescription": "The Amazon Simple Storage Service \\(Amazon S3\\) URI or local file path of the state machine definition written in the [Amazon States Language](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-amazon-states-language.html)\\.  \nIf you provide a local file path, the template must go through the workflow that includes the `sam deploy` or `sam package` command to correctly transform the definition\\. To do this, you must use version 0\\.52\\.0 or later of the AWS SAM CLI\\.  \nYou must provide either a `Definition` or a `DefinitionUri`\\.  \n*Type*: String \\| [S3Location](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-definitions3location)  \n*Required*: Conditional  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`DefinitionS3Location`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-definitions3location) property of an `AWS::StepFunctions::StateMachine` resource\\.",
           "title": "DefinitionUri"
         },
+        "DeploymentPreference": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
         "Events": {
           "additionalProperties": {
             "anyOf": [

schema_source/sam.schema.json

@@ -7102,6 +7102,9 @@
     "samtranslator__internal__schema_source__aws_serverless_statemachine__Properties": {
       "additionalProperties": false,
       "properties": {
+        "AutoPublishAlias": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
         "Definition": {
           "markdownDescription": "The state machine definition is an object, where the format of the object matches the format of your AWS SAM template file, for example, JSON or YAML\\. State machine definitions adhere to the [Amazon States Language](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-amazon-states-language.html)\\.  \nFor an example of an inline state machine definition, see [Examples](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/#sam-resource-statemachine--examples.html#sam-resource-statemachine--examples)\\.  \nYou must provide either a `Definition` or a `DefinitionUri`\\.  \n*Type*: Map  \n*Required*: Conditional  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
           "title": "Definition",
@@ -7124,6 +7127,9 @@
           "markdownDescription": "The Amazon Simple Storage Service \\(Amazon S3\\) URI or local file path of the state machine definition written in the [Amazon States Language](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-amazon-states-language.html)\\.  \nIf you provide a local file path, the template must go through the workflow that includes the `sam deploy` or `sam package` command to correctly transform the definition\\. To do this, you must use version 0\\.52\\.0 or later of the AWS SAM CLI\\.  \nYou must provide either a `Definition` or a `DefinitionUri`\\.  \n*Type*: String \\| [S3Location](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-definitions3location)  \n*Required*: Conditional  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`DefinitionS3Location`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-definitions3location) property of an `AWS::StepFunctions::StateMachine` resource\\.",
           "title": "DefinitionUri"
         },
+        "DeploymentPreference": {
+          "$ref": "#/definitions/PassThroughProp"
+        },
         "Events": {
           "additionalProperties": {
             "anyOf": [

tests/model/stepfunctions/test_state_machine_generator.py

@@ -28,7 +28,9 @@ def setUp(self):
             "event_resolver": None,
             "tags": None,
             "resource_attributes": None,
-            "passthrough_resource_attributes": None,
+            "passthrough_resource_attributes": {},
+            "auto_publish_alias": None,
+            "deployment_preference": None,
         }
 
     def test_state_machine_no_definition_source(self):
@@ -145,3 +147,43 @@ def test_state_machine_with_unsupported_event_source(self):
         self.kwargs["event_resources"] = {"KinesesEvent": {}}
         with self.assertRaises(InvalidEventException):
             StateMachineGenerator(**self.kwargs).to_cloudformation()
+
+    def test_state_machine_with_managed_traffic_shifting_properties(self):
+        self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
+        self.kwargs["role"] = "my-test-role-arn"
+        self.kwargs["auto_publish_alias"] = "live"
+        self.kwargs["deployment_preference"] = {"Type": "ALL_AT_ONCE"}
+        generated_managed_traffic_shifting_resources = StateMachineGenerator(
+            **self.kwargs
+        )._generate_managed_traffic_shifting_resources()
+        self.assertEqual(
+            generated_managed_traffic_shifting_resources[0].resource_type, "AWS::StepFunctions::StateMachineVersion"
+        )
+        self.assertEqual(
+            generated_managed_traffic_shifting_resources[1].resource_type, "AWS::StepFunctions::StateMachineAlias"
+        )
+
+    def test_state_machine_with_auto_publish_alias_and_no_deployment_preference(self):
+        self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
+        self.kwargs["role"] = "my-test-role-arn"
+        self.kwargs["auto_publish_alias"] = "live"
+        generated_managed_traffic_shifting_resources = StateMachineGenerator(
+            **self.kwargs
+        )._generate_managed_traffic_shifting_resources()
+        self.assertEqual(
+            generated_managed_traffic_shifting_resources[0].resource_type, "AWS::StepFunctions::StateMachineVersion"
+        )
+        self.assertEqual(
+            generated_managed_traffic_shifting_resources[1].resource_type, "AWS::StepFunctions::StateMachineAlias"
+        )
+
+    def test_state_machine_with_deployment_preference_and_no_auto_publish_alias(self):
+        self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
+        self.kwargs["role"] = "my-test-role-arn"
+        self.kwargs["deployment_preference"] = {"Type": "ALL_AT_ONCE"}
+        with self.assertRaises(InvalidResourceException) as error:
+            StateMachineGenerator(**self.kwargs).to_cloudformation()
+        self.assertEqual(
+            error.exception.message,
+            "Resource with id [StateMachineId] is invalid. 'DeploymentPreference' requires 'AutoPublishAlias' property to be specified.",
+        )