Source code for sceptre.hooks.asg_scaling_processes

# -*- coding: utf-8 -*-
from six import string_types

from sceptre.hooks import Hook
from sceptre.exceptions import InvalidHookArgumentTypeError
from sceptre.exceptions import InvalidHookArgumentSyntaxError
from sceptre.exceptions import InvalidHookArgumentValueError

[docs]class ASGScalingProcesses(Hook): """ Resumes or suspends autoscaling group scaling processes. This is useful as scheduled actions must be suspended when updating stacks with autoscaling groups. """ def __init__(self, *args, **kwargs): super(ASGScalingProcesses, self).__init__(*args, **kwargs)
[docs] def run(self): """ Either suspends or resumes any scaling processes on all autoscaling groups within the current stack. :raises: InvalidHookArgumentSyntaxError, when syntax is not using "::". :raises: InvalidHookArgumentTypeError, if argument is not a string. :raises: InvalidHookArgumentValueError, if not using resume or suspend. """ if not isinstance(self.argument, string_types): raise InvalidHookArgumentTypeError( 'The argument "{0}" is the wrong type - asg_scaling_processes ' "hooks require arguments of type string.".format(self.argument) ) if "::" not in str(self.argument): raise InvalidHookArgumentSyntaxError( 'Wrong syntax for the argument "{0}" - asg_scaling_processes ' "hooks use:" "- !asg_scaling_processes <suspend|resume>::<process-name>".format( self.argument ) ) action, scaling_processes = self.argument.split("::") if action not in ["resume", "suspend"]: raise InvalidHookArgumentValueError( 'The argument "{0}" is invalid - valid arguments for ' 'asg_scaling_processes hooks are "resume" or "suspend".'.format(action) ) action += "_processes" autoscaling_group_names = self._find_autoscaling_groups() for autoscaling_group in autoscaling_group_names: service="autoscaling", command=action, kwargs={ "AutoScalingGroupName": autoscaling_group, "ScalingProcesses": [scaling_processes], }, )
def _get_stack_resources(self): """ Retrieves all resources in stack. :return: list """ response = service="cloudformation", command="describe_stack_resources", kwargs={"StackName": self.stack.external_name}, ) return response.get("StackResources", []) def _find_autoscaling_groups(self): """ Retrieves all the autoscaling groups :return: list [str] """ asg_names = [] resources = self._get_stack_resources() resource_type = "AWS::AutoScaling::AutoScalingGroup" for resource in resources: if resource.get("ResourceType", False) == resource_type: asg_names.append(resource["PhysicalResourceId"]) return asg_names