""" Cloudformation deploy class which also streams events and changeset information This was ported over from the sam-cli repo """ # Copyright 2012-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # http://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. # This is a modified version of the Deployer class from aws-sam-cli # (and its dependencies) to work with python 2 # Modifications: # - Imports now reference local classes # - Alternative imports for python2 # - py3 -> py2 migrations (ex: "".format() instead of f"", no "from" for raise) # - Moved UserException to exceptions.py # - Moved DeployColor to colors.py # - Removed unnecessary functions from artifact_exporter import logging import sys import time from collections import OrderedDict from datetime import datetime, timezone import botocore from integration.helpers.deployer.exceptions import exceptions as deploy_exceptions from integration.helpers.deployer.utils.artifact_exporter import mktempfile from integration.helpers.deployer.utils.colors import DeployColor from integration.helpers.deployer.utils.retry import retry_with_exponential_backoff_and_jitter from integration.helpers.deployer.utils.table_print import ( MIN_OFFSET, newline_per_item, pprint_column_names, pprint_columns, ) from integration.helpers.deployer.utils.time_util import utc_to_timestamp from integration.helpers.resource import generate_suffix LOG = logging.getLogger(__name__) DESCRIBE_STACK_EVENTS_FORMAT_STRING = ( "{ResourceStatus:<{0}} {ResourceType:<{1}} {LogicalResourceId:<{2}} {ResourceStatusReason:<{3}}" ) DESCRIBE_STACK_EVENTS_DEFAULT_ARGS = OrderedDict( { "ResourceStatus": "ResourceStatus", "ResourceType": "ResourceType", "LogicalResourceId": "LogicalResourceId", "ResourceStatusReason": "ResourceStatusReason", } ) DESCRIBE_STACK_EVENTS_TABLE_HEADER_NAME = "CloudFormation events from changeset" DESCRIBE_CHANGESET_FORMAT_STRING = "{Operation:<{0}} {LogicalResourceId:<{1}} {ResourceType:<{2}} {Replacement:<{3}}" DESCRIBE_CHANGESET_DEFAULT_ARGS = OrderedDict( { "Operation": "Operation", "LogicalResourceId": "LogicalResourceId", "ResourceType": "ResourceType", "Replacement": "Replacement", } ) DESCRIBE_CHANGESET_TABLE_HEADER_NAME = "CloudFormation stack changeset" OUTPUTS_FORMAT_STRING = "{Outputs:<{0}}" OUTPUTS_DEFAULTS_ARGS = OrderedDict({"Outputs": "Outputs"}) OUTPUTS_TABLE_HEADER_NAME = "CloudFormation outputs from deployed stack" class Deployer: def __init__(self, cloudformation_client, changeset_prefix="sam-integ-"): self._client = cloudformation_client self.changeset_prefix = changeset_prefix # 500ms of sleep time between stack checks and describe stack events. self.client_sleep = 0.5 # 2000ms of backoff time which is exponentially used, when there are exceptions during describe stack events self.backoff = 2 # Maximum number of attempts before raising exception back up the chain. self.max_attempts = 3 self.deploy_color = DeployColor() def create_changeset( self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags, changeset_type, ): """ Call Cloudformation to create a changeset and wait for it to complete :param stack_name: Name or ID of stack :param cfn_template: CloudFormation template string :param parameter_values: Template parameters object :param capabilities: Array of capabilities passed to CloudFormation :param tags: Array of tags passed to CloudFormation :param tags: the type of the changeset :return: """ if type == "UPDATE": # UsePreviousValue not valid if parameter is new summary = self._client.get_template_summary(StackName=stack_name) existing_parameters = [parameter["ParameterKey"] for parameter in summary["Parameters"]] parameter_values = [ x for x in parameter_values if not (x.get("UsePreviousValue", False) and x["ParameterKey"] not in existing_parameters) ] else: # When creating a new stack, UsePreviousValue=True is invalid. # For such parameters, users should either override with new value, # or set a Default value in template to successfully create a stack. parameter_values = [x for x in parameter_values if not x.get("UsePreviousValue", False)] # Each changeset will get a unique name based on time. # Description is also setup based on current date and that SAM CLI is used. kwargs = { "ChangeSetName": self.changeset_prefix + str(int(time.time())), "StackName": stack_name, "TemplateBody": cfn_template, "ChangeSetType": changeset_type, "Parameters": parameter_values, "Capabilities": capabilities, "Description": f"Created by SAM CLI at {datetime.now(timezone.utc).isoformat()} UTC", "Tags": tags, } # If an S3 uploader is available, use TemplateURL to deploy rather than # TemplateBody. This is required for large templates. if s3_uploader: with mktempfile() as temporary_file: temporary_file.write(kwargs.pop("TemplateBody")) temporary_file.flush() # add random suffix to file_name to enable multiple tests run in the same time template_file_name = "template" + generate_suffix() # TemplateUrl property requires S3 URL to be in path-style format s3_uploader.upload_file(template_file_name, temporary_file.name) kwargs["TemplateURL"] = s3_uploader.get_s3_uri(template_file_name) # don't set these arguments if not specified to use existing values if role_arn is not None: kwargs["RoleARN"] = role_arn if notification_arns is not None: kwargs["NotificationARNs"] = notification_arns return self._create_change_set(stack_name=stack_name, changeset_type=changeset_type, **kwargs) def _create_change_set(self, stack_name, changeset_type, **kwargs): try: resp = self._client.create_change_set(**kwargs) return resp, changeset_type except botocore.exceptions.ClientError as ex: if "The bucket you are attempting to access must be addressed using the specified endpoint" in str(ex): raise deploy_exceptions.DeployBucketInDifferentRegionError( f"Failed to create/update stack {stack_name}" ) raise deploy_exceptions.ChangeSetError(stack_name=stack_name, msg=str(ex)) except Exception as ex: LOG.debug("Unable to create changeset", exc_info=ex) raise deploy_exceptions.ChangeSetError(stack_name=stack_name, msg=str(ex)) @pprint_column_names( format_string=DESCRIBE_CHANGESET_FORMAT_STRING, format_kwargs=DESCRIBE_CHANGESET_DEFAULT_ARGS, table_header=DESCRIBE_CHANGESET_TABLE_HEADER_NAME, ) def describe_changeset(self, change_set_id, stack_name, **kwargs): """ Call Cloudformation to describe a changeset :param change_set_id: ID of the changeset :param stack_name: Name of the CloudFormation stack :return: dictionary of changes described in the changeset. """ paginator = self._client.get_paginator("describe_change_set") response_iterator = paginator.paginate(ChangeSetName=change_set_id, StackName=stack_name) changes = {"Add": [], "Modify": [], "Remove": []} changes_showcase = {"Add": "+ Add", "Modify": "* Modify", "Remove": "- Delete"} changeset = False for item in response_iterator: cf_changes = item.get("Changes") for change in cf_changes: changeset = True resource_props = change.get("ResourceChange") action = resource_props.get("Action") 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") ), } ) for k, v in changes.items(): for value in v: row_color = self.deploy_color.get_changeset_action_color(action=k) pprint_columns( columns=[ changes_showcase.get(k, k), value["LogicalResourceId"], value["ResourceType"], value["Replacement"], ], width=kwargs["width"], margin=kwargs["margin"], format_string=DESCRIBE_CHANGESET_FORMAT_STRING, format_args=kwargs["format_args"], columns_dict=DESCRIBE_CHANGESET_DEFAULT_ARGS.copy(), 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(), ) return changes def wait_for_changeset(self, changeset_id, stack_name): """ Waits until the changeset creation completes :param changeset_id: ID or name of the changeset :param stack_name: Stack name :return: Latest status of the create-change-set operation """ sys.stdout.write("\nWaiting for changeset to be created...\n") sys.stdout.flush() # Wait for changeset to be created waiter = self._client.get_waiter("change_set_create_complete") # Poll every 5 seconds. Changeset creation should be fast waiter_config = {"Delay": 5} try: waiter.wait(ChangeSetName=changeset_id, StackName=stack_name, WaiterConfig=waiter_config) except botocore.exceptions.WaiterError as ex: LOG.error("Waiter exception waiting for changeset", exc_info=ex) resp = ex.last_response status = resp.get("Status", "") reason = resp.get("StatusReason", "") if ( status == "FAILED" and "The submitted information didn't contain changes." in reason or "No updates are to be performed" in reason ): raise deploy_exceptions.ChangeEmptyError(stack_name=stack_name) raise deploy_exceptions.ChangeSetError( stack_name=stack_name, msg=f"ex: {ex} Status: {status}. Reason: {reason}" ) def execute_changeset(self, changeset_id, stack_name): """ Calls CloudFormation to execute changeset :param changeset_id: ID of the changeset :param stack_name: Name or ID of the stack :return: Response from execute-change-set call """ try: return self._client.execute_change_set(ChangeSetName=changeset_id, StackName=stack_name) except botocore.exceptions.ClientError as ex: raise deploy_exceptions.DeployFailedError(stack_name=stack_name, msg=str(ex)) def get_last_event_time(self, stack_name): """ Finds the last event time stamp thats present for the stack, if not get the current time :param stack_name: Name or ID of the stack :return: unix epoch """ try: return utc_to_timestamp( self._client.describe_stack_events(StackName=stack_name)["StackEvents"][0]["Timestamp"] ) except KeyError: return time.time() def _check_stack_complete(self, status): return "COMPLETE" in status and "CLEANUP" not in status def wait_for_execute(self, stack_name, changeset_type): sys.stdout.write( "\n{} - Waiting for stack create/update " "to complete\n".format(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) ) sys.stdout.flush() # Pick the right waiter if changeset_type == "CREATE": waiter = self._client.get_waiter("stack_create_complete") elif changeset_type == "UPDATE": waiter = self._client.get_waiter("stack_update_complete") else: raise RuntimeError(f"Invalid changeset type {changeset_type}") # Poll every 30 seconds. Polling too frequently risks hitting rate limits # on CloudFormation's DescribeStacks API waiter_config = {"Delay": 30, "MaxAttempts": 120} self._wait(stack_name, waiter, waiter_config) outputs = self.get_stack_outputs(stack_name=stack_name, echo=False) if outputs: self._display_stack_outputs(outputs) @retry_with_exponential_backoff_and_jitter(deploy_exceptions.ThrottlingError, 5, 360) def _wait(self, stack_name, waiter, waiter_config): try: waiter.wait(StackName=stack_name, WaiterConfig=waiter_config) except botocore.exceptions.WaiterError as ex: LOG.debug("Execute changeset waiter exception", exc_info=ex) if "Throttling" in str(ex): raise deploy_exceptions.ThrottlingError(stack_name=stack_name, msg=str(ex)) else: raise deploy_exceptions.DeployFailedError(stack_name=stack_name, msg=str(ex)) def create_and_wait_for_changeset( self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags, changeset_type, ): try: result, changeset_type = self.create_changeset( stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags, changeset_type, ) self.wait_for_changeset(result["Id"], stack_name) self.describe_changeset(result["Id"], stack_name) return result except deploy_exceptions.ChangeEmptyError: return {} except botocore.exceptions.ClientError as ex: raise deploy_exceptions.DeployFailedError(stack_name=stack_name, msg=str(ex)) @pprint_column_names( format_string=OUTPUTS_FORMAT_STRING, format_kwargs=OUTPUTS_DEFAULTS_ARGS, table_header=OUTPUTS_TABLE_HEADER_NAME ) def _display_stack_outputs(self, stack_outputs, **kwargs): for counter, output in enumerate(stack_outputs): for k, v in [ ("Key", output.get("OutputKey")), ("Description", output.get("Description", "-")), ("Value", output.get("OutputValue")), ]: pprint_columns( columns=[f"{k:<{MIN_OFFSET}}{v:<{MIN_OFFSET}}"], width=kwargs["width"], margin=kwargs["margin"], format_string=OUTPUTS_FORMAT_STRING, format_args=kwargs["format_args"], columns_dict=OUTPUTS_DEFAULTS_ARGS.copy(), color="green", replace_whitespace=False, break_long_words=False, drop_whitespace=False, ) newline_per_item(stack_outputs, counter) @retry_with_exponential_backoff_and_jitter(deploy_exceptions.ThrottlingError, 5, 360) def get_stack_outputs(self, stack_name, echo=True): try: stacks_description = self._client.describe_stacks(StackName=stack_name) try: outputs = stacks_description["Stacks"][0]["Outputs"] if echo: sys.stdout.write(f"\nStack {stack_name} outputs:\n") sys.stdout.flush() self._display_stack_outputs(stack_outputs=outputs) return outputs except KeyError: return None except botocore.exceptions.ClientError as ex: if "Throttling" in str(ex): raise deploy_exceptions.ThrottlingError(stack_name=stack_name, msg=str(ex)) else: raise deploy_exceptions.DeployStackOutPutFailedError(stack_name=stack_name, msg=str(ex))