-
Notifications
You must be signed in to change notification settings - Fork 311
/
asg_scaling_processes.py
89 lines (75 loc) · 3.18 KB
/
asg_scaling_processes.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
# -*- 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
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)
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:
self.stack.connection_manager.call(
service="autoscaling",
command=action,
kwargs={
"AutoScalingGroupName": autoscaling_group,
"ScalingProcesses": [scaling_processes]
}
)
def _get_stack_resources(self):
"""
Retrieves all resources in stack.
:return: list
"""
response = self.stack.connection_manager.call(
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