feat: Add nested stack changeset support to sam deploy

samcli/commands/deploy/command.py

@@ -108,6 +108,15 @@
     is_flag=True,
     help="Prompt to confirm if the computed changeset is to be deployed by SAM CLI.",
 )
+@click.option(
+    "--include-nested-stacks/--no-include-nested-stacks",
+    default=True,
+    required=False,
+    is_flag=True,
+    help="Display changes for nested stacks in the changeset. "
+    "For large nested stack hierarchies, use --no-include-nested-stacks to reduce output verbosity. "
+    "Defaults to displaying nested stack changes.",
+)
 @click.option(
     "--disable-rollback/--no-disable-rollback",
     default=False,
@@ -191,6 +200,7 @@ def cli(
     metadata,
     guided,
     confirm_changeset,
+    include_nested_stacks,
     signing_profiles,
     resolve_s3,
     resolve_image_repos,
@@ -226,6 +236,7 @@ def cli(
         metadata,
         guided,
         confirm_changeset,
+        include_nested_stacks,
         ctx.region,
         ctx.profile,
         signing_profiles,
@@ -260,6 +271,7 @@ def do_cli(
     metadata,
     guided,
     confirm_changeset,
+    include_nested_stacks,
     region,
     profile,
     signing_profiles,
@@ -300,6 +312,7 @@ def do_cli(
             config_env=config_env,
             config_file=config_file,
             disable_rollback=disable_rollback,
+            include_nested_stacks=include_nested_stacks,
         )
         guided_context.run()
     else:
@@ -370,6 +383,7 @@ def do_cli(
             region=guided_context.guided_region if guided else region,
             profile=profile,
             confirm_changeset=guided_context.confirm_changeset if guided else confirm_changeset,
+            include_nested_stacks=include_nested_stacks,
             signing_profiles=guided_context.signing_profiles if guided else signing_profiles,
             use_changeset=True,
             disable_rollback=guided_context.disable_rollback if guided else disable_rollback,

samcli/commands/deploy/core/options.py

@@ -34,6 +34,7 @@
     "no_execute_changeset",
     "fail_on_empty_changeset",
     "confirm_changeset",
+    "include_nested_stacks",
     "disable_rollback",
     "on_failure",
     "force_upload",

samcli/commands/deploy/deploy_context.py

@@ -69,12 +69,13 @@ def __init__(
         region,
         profile,
         confirm_changeset,
-        signing_profiles,
-        use_changeset,
-        disable_rollback,
-        poll_delay,
-        on_failure,
-        max_wait_duration,
+        include_nested_stacks=True,
+        signing_profiles=None,
+        use_changeset=True,
+        disable_rollback=False,
+        poll_delay=0.5,
+        on_failure=None,
+        max_wait_duration=60,
     ):
         self.template_file = template_file
         self.stack_name = stack_name
@@ -101,6 +102,7 @@ def __init__(
         self.s3_uploader = None
         self.deployer = None
         self.confirm_changeset = confirm_changeset
+        self.include_nested_stacks = include_nested_stacks
         self.signing_profiles = signing_profiles
         self.use_changeset = use_changeset
         self.disable_rollback = disable_rollback
@@ -257,6 +259,7 @@ def deploy(
                     notification_arns=notification_arns,
                     s3_uploader=s3_uploader,
                     tags=tags,
+                    include_nested_stacks=self.include_nested_stacks,
                 )
                 click.echo(self.MSG_SHOWCASE_CHANGESET.format(changeset_id=result["Id"]))
 

samcli/commands/deploy/guided_context.py

@@ -62,6 +62,7 @@ def __init__(
         config_env=None,
         config_file=None,
         disable_rollback=None,
+        include_nested_stacks=True,
     ):
         self.template_file = template_file
         self.stack_name = stack_name
@@ -95,6 +96,7 @@ def __init__(
         self.color = Colored()
         self.function_provider = None
         self.disable_rollback = disable_rollback
+        self.include_nested_stacks = include_nested_stacks
 
     @property
     def guided_capabilities(self):
@@ -584,6 +586,7 @@ def run(self):
                 region=self.guided_region,
                 profile=self.guided_profile,
                 confirm_changeset=self.confirm_changeset,
+                include_nested_stacks=self.include_nested_stacks,
                 capabilities=self._capabilities,
                 signing_profiles=self.signing_profiles,
                 disable_rollback=self.disable_rollback,

samcli/commands/sync/command.py

@@ -392,6 +392,7 @@ def do_cli(
                     no_execute_changeset=True,
                     fail_on_empty_changeset=True,
                     confirm_changeset=False,
+                    include_nested_stacks=True,
                     use_changeset=False,
                     force_upload=True,
                     signing_profiles=None,

samcli/lib/deploy/deployer.py

@@ -17,13 +17,15 @@
 
 import logging
 import math
+import re
 import sys
 import time
 from collections import OrderedDict, deque
 from datetime import datetime, timezone
 from typing import Dict, List, Optional
 
 import botocore
+import click
 
 from samcli.commands._utils.table_print import MIN_OFFSET, newline_per_item, pprint_column_names, pprint_columns
 from samcli.commands.deploy import exceptions as deploy_exceptions
@@ -141,7 +143,16 @@ def has_stack(self, stack_name):
             raise e
 
     def create_changeset(
-        self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+        self,
+        stack_name,
+        cfn_template,
+        parameter_values,
+        capabilities,
+        role_arn,
+        notification_arns,
+        s3_uploader,
+        tags,
+        include_nested_stacks=True,
     ):
         """
         Call Cloudformation to create a changeset and wait for it to complete
@@ -154,6 +165,7 @@ def create_changeset(
         :param notification_arns: Arns for sending notifications
         :param s3_uploader: S3Uploader object to upload files to S3 buckets
         :param tags: Array of tags passed to CloudFormation
+        :param include_nested_stacks: Whether to include nested stack changes in changeset (default: True)
         :return:
         """
         if not self.has_stack(stack_name):
@@ -183,6 +195,7 @@ def create_changeset(
             "Parameters": parameter_values,
             "Description": "Created by SAM CLI at {0} UTC".format(datetime.now(timezone.utc).isoformat()),
             "Tags": tags,
+            "IncludeNestedStacks": include_nested_stacks,
         }
 
         kwargs = self._process_kwargs(kwargs, s3_uploader, capabilities, role_arn, notification_arns)
@@ -243,27 +256,69 @@ def describe_changeset(self, change_set_id, stack_name, **kwargs):
         :param kwargs: Other arguments to pass to pprint_columns()
         :return: dictionary of changes described in the changeset.
         """
+        # Display changes for parent stack first
+        changeset = self._display_changeset_changes(change_set_id, stack_name, is_parent=True, **kwargs)
+
+        if changeset is None:
+            # There can be cases where there are no changes,
+            # but could be an an addition of a SNS notification topic.
+            pprint_columns(
+                columns=["-", "-", "-", "-"],
+                width=kwargs["width"],
+                margin=kwargs["margin"],
+                format_string=DESCRIBE_CHANGESET_FORMAT_STRING,
+                format_args=kwargs["format_args"],
+                columns_dict=DESCRIBE_CHANGESET_DEFAULT_ARGS.copy(),
+            )
+            return {"Add": [], "Modify": [], "Remove": []}
+
+        return changeset
+
+    def _display_changeset_changes(
+        self, change_set_id: str, stack_name: str, is_parent: bool = False, **kwargs
+    ) -> Optional[Dict[str, List]]:
+        """
+        Display changes for a changeset, including nested stack changes recursively
+
+        :param change_set_id: ID of the changeset
+        :param stack_name: Name of the CloudFormation stack
+        :param is_parent: Whether this is the parent stack (used to control header display)
+        :param kwargs: Other arguments to pass to pprint_columns()
+        :return: dictionary of changes or None if no changes
+        """
         paginator = self._client.get_paginator("describe_change_set")
         response_iterator = paginator.paginate(ChangeSetName=change_set_id, StackName=stack_name)
-        changes = {"Add": [], "Modify": [], "Remove": []}
+        changes: Dict[str, List] = {"Add": [], "Modify": [], "Remove": []}
         changes_showcase = {"Add": "+ Add", "Modify": "* Modify", "Remove": "- Delete"}
-        changeset = False
+        changeset_found = False
+        nested_changesets = []
+
         for item in response_iterator:
-            cf_changes = item.get("Changes")
+            cf_changes = item.get("Changes", [])
             for change in cf_changes:
-                changeset = True
-                resource_props = change.get("ResourceChange")
+                changeset_found = True
+                resource_props = change.get("ResourceChange", {})
                 action = resource_props.get("Action")
+                resource_type = resource_props.get("ResourceType")
+                logical_id = resource_props.get("LogicalResourceId")
+
+                # Check if this is a nested stack with its own changeset
+                nested_changeset_id = resource_props.get("ChangeSetId")
+                if resource_type == "AWS::CloudFormation::Stack" and nested_changeset_id:
+                    nested_changesets.append(
+                        {"changeset_id": nested_changeset_id, "logical_id": logical_id, "action": action}
+                    )
+
+                replacement = resource_props.get("Replacement")
                 changes[action].append(
                     {
-                        "LogicalResourceId": resource_props.get("LogicalResourceId"),
-                        "ResourceType": resource_props.get("ResourceType"),
-                        "Replacement": (
-                            "N/A" if resource_props.get("Replacement") is None else resource_props.get("Replacement")
-                        ),
+                        "LogicalResourceId": logical_id,
+                        "ResourceType": resource_type,
+                        "Replacement": "N/A" if replacement is None else replacement,
                     }
                 )
 
+        # Display changes for this stack
         for k, v in changes.items():
             for value in v:
                 row_color = self.deploy_color.get_changeset_action_color(action=k)
@@ -282,19 +337,27 @@ def describe_changeset(self, change_set_id, stack_name, **kwargs):
                     color=row_color,
                 )
 
-        if not changeset:
-            # There can be cases where there are no changes,
-            # but could be an an addition of a SNS notification topic.
-            pprint_columns(
-                columns=["-", "-", "-", "-"],
-                width=kwargs["width"],
-                margin=kwargs["margin"],
-                format_string=DESCRIBE_CHANGESET_FORMAT_STRING,
-                format_args=kwargs["format_args"],
-                columns_dict=DESCRIBE_CHANGESET_DEFAULT_ARGS.copy(),
-            )
+        # Recursively display nested stack changes
+        # Only process nested stacks when is_parent=True to avoid duplicates
+        if is_parent:
+            for nested in nested_changesets:
+                try:
+                    # Display nested stack header
+                    click.echo(f"\n[Nested Stack: {nested['logical_id']}]")
+
+                    # Get the stack name from the changeset to support recursive call
+                    nested_response = self._client.describe_change_set(ChangeSetName=nested["changeset_id"])
+                    nested_stack_name = nested_response.get("StackName")
+                    if nested_stack_name:
+                        # Recursively call to display nested changes (supports arbitrary nesting depth)
+                        self._display_changeset_changes(
+                            nested["changeset_id"], nested_stack_name, is_parent=True, **kwargs
+                        )
+                except Exception as e:
+                    LOG.debug("Failed to describe nested changeset %s: %s", nested["changeset_id"], e)
+                    click.echo(f"Unable to fetch changes: {str(e)}")
 
-        return changes
+        return changes if changeset_found else None
 
     def wait_for_changeset(self, changeset_id, stack_name):
         """
@@ -330,8 +393,49 @@ def wait_for_changeset(self, changeset_id, stack_name):
             ):
                 raise deploy_exceptions.ChangeEmptyError(stack_name=stack_name)
 
+            # Check if this is a nested stack changeset error
+            if status == "FAILED" and "Nested change set" in reason:
+                # Try to fetch detailed error from nested changeset
+                detailed_error = self._get_nested_changeset_error(reason)
+                if detailed_error:
+                    reason = detailed_error
+
             raise ChangeSetError(stack_name=stack_name, msg=f"ex: {ex} Status: {status}. Reason: {reason}") from ex
 
+    def _get_nested_changeset_error(self, status_reason: str) -> Optional[str]:
+        """
+        Extract and fetch detailed error from nested changeset
+
+        :param status_reason: The status reason from parent changeset
+        :return: Detailed error message or None
+        """
+        try:
+            # Extract nested changeset ARN from status reason
+            # Format: "Nested change set arn:aws:cloudformation:... was not successfully created: Currently in FAILED."
+            # Support all AWS partitions: aws, aws-cn, aws-us-gov, aws-iso, aws-iso-b
+            match = re.search(
+                r"arn:aws[-a-z]*:cloudformation:[^:]+:[^:]+:changeSet/([^/]+)/([a-f0-9-]+)", status_reason
+            )
+            if match:
+                nested_changeset_arn = match.group(0)
+
+                # Fetch nested changeset details to get the actual stack name
+                try:
+                    response = self._client.describe_change_set(ChangeSetName=nested_changeset_arn)
+                    nested_stack_name = response.get("StackName")
+                    nested_status = response.get("Status")
+                    nested_reason = response.get("StatusReason", "")
+
+                    if nested_status == "FAILED" and nested_reason and nested_stack_name:
+                        return f"Nested stack '{nested_stack_name}' changeset failed: {nested_reason}"
+                except Exception as e:
+                    LOG.debug("Failed to fetch nested changeset details: %s", e)
+
+        except Exception as e:
+            LOG.debug("Failed to parse nested changeset error: %s", e)
+
+        return None
+
     def execute_changeset(self, changeset_id, stack_name, disable_rollback):
         """
         Calls CloudFormation to execute changeset
@@ -556,11 +660,28 @@ def wait_for_execute(
                 raise ex
 
     def create_and_wait_for_changeset(
-        self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+        self,
+        stack_name,
+        cfn_template,
+        parameter_values,
+        capabilities,
+        role_arn,
+        notification_arns,
+        s3_uploader,
+        tags,
+        include_nested_stacks=True,
     ):
         try:
             result, changeset_type = self.create_changeset(
-                stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+                stack_name,
+                cfn_template,
+                parameter_values,
+                capabilities,
+                role_arn,
+                notification_arns,
+                s3_uploader,
+                tags,
+                include_nested_stacks,
             )
             self.wait_for_changeset(result["Id"], stack_name)
             self.describe_changeset(result["Id"], stack_name)