in source/code/builders/action_template_builder.py [0:0]
def _setup_common_parameters(self):
"""
Setup non action specific parameters
:return:
"""
def setup_region_parameter():
"""
Create the region parameter
:return:
"""
if self.has_regions_parameter:
self._template_parameters[configuration.CONFIG_REGIONS] = {
"Type": "CommaDelimitedList",
"Default": self.region,
"Description": PARAM_DESCRIPTION_REGIONS.format(self.region)
}
self._parameter_labels[configuration.CONFIG_REGIONS] = {"default": PARAM_LABEL_REGIONS}
self._parameter_groups[0]["Parameters"].insert(2, configuration.CONFIG_REGIONS)
def setup_cross_account_parameters():
"""
Creates cross account parameter
:return:
"""
if self.action_properties.get(actions.ACTION_CROSS_ACCOUNT, True):
self._template_parameters[configuration.CONFIG_THIS_ACCOUNT] = {
"Type": "String",
"AllowedValues": YES_NO,
"Default": YES,
"Description": PARAM_DESCRIPTION_THIS_ACCOUNT.format(services.get_aws_account())
}
order = 3 if self.is_regional_service else 2
self._parameter_labels[configuration.CONFIG_THIS_ACCOUNT] = {"default": PARAM_LABEL_THIS_ACCOUNT}
self._parameter_groups[0]["Parameters"].insert(order, configuration.CONFIG_THIS_ACCOUNT)
self._template_parameters[configuration.CONFIG_ACCOUNTS] = {
"Type": "CommaDelimitedList",
"Description": PARAM_DESCRIPTION_ACCOUNTS
}
self._parameter_labels[configuration.CONFIG_ACCOUNTS] = {"default": PARAM_LABEL_ACCOUNTS}
self._parameter_groups[0]["Parameters"].insert(order + 1, configuration.CONFIG_ACCOUNTS)
self._template_parameters[configuration.CONFIG_TASK_CROSS_ACCOUNT_ROLE_NAME] = {
"Type": "String",
"Description": PARAM_DESCRIPTION_CROSS_ACCOUNT_ROLE_NAME.format(handlers.default_rolename_for_stack(),
self.aws_account,
self.ops_automator_role.split("/")[-1],
self.automator_stack_name)
}
self._parameter_labels[configuration.CONFIG_TASK_CROSS_ACCOUNT_ROLE_NAME] = {
"default": PARAM_LABEL_CROSS_ACCOUNT_ROLE_NAME
}
self._parameter_groups[0]["Parameters"].insert(order + 2, configuration.CONFIG_TASK_CROSS_ACCOUNT_ROLE_NAME)
def setup_timeout_parameter():
"""
Creates a timeout parameter if the task has a completion check method
:return:
"""
if not self.has_completion_logic:
return
timeout = self.action_properties.get(actions.ACTION_COMPLETION_TIMEOUT_MINUTES,
actions.DEFAULT_COMPLETION_TIMEOUT_MINUTES_DEFAULT)
self._template_parameters[configuration.CONFIG_TASK_TIMEOUT] = {
"Type": "Number",
"MinValue": 1,
"Default": str(timeout),
"Description": PARAM_DESCRIPTION_TIMEOUT
}
self._parameter_labels[configuration.CONFIG_TASK_TIMEOUT] = {"default": PARAM_LABEL_TIMEOUT}
self._parameter_groups[0]["Parameters"].insert(2, configuration.CONFIG_TASK_TIMEOUT)
def setup_tag_filter_parameter():
if self.action_properties.get(actions.ACTION_NO_TAG_SELECT, False):
return
# test if the resource/service supports tags
action_resources = self.action_properties.get(actions.ACTION_RESOURCES, "")
service = self.action_properties[actions.ACTION_SERVICE]
service_resource_with_tags = services.create_service(service).resources_with_tags
if action_resources == "":
self._resource_supports_tags = len(service_resource_with_tags) != 0
else:
self._resource_supports_tags = action_resources.lower() in [r.lower() for r in service_resource_with_tags]
if not self._resource_supports_tags:
return
# task tag filter
self._template_parameters[configuration.CONFIG_TAG_FILTER] = {
"Type": "String",
"Description": PARAM_DESCRIPTION_TAG_FILTER.format(self.automator_tag_name)
}
if not (self.action_properties.get(actions.ACTION_ALLOW_TAGFILTER_WILDCARD, True)):
self._template_parameters[configuration.CONFIG_TAG_FILTER]["Description"] += PARAM_DESCRIPTION_TAG_FILER_NO_WILDCARD
self._parameter_labels[configuration.CONFIG_TAG_FILTER] = {"default": PARAM_LABEL_FILTER}
self._parameter_groups[0]["Parameters"].insert(2, configuration.CONFIG_TAG_FILTER)
def build_memory_parameter(size_group, lambda_size, lambda_size_param, config_ecs_memory_param, description, label,
ecs_memory_label, ecs_description):
memory_settings = self.ops_automator_stack_template["Mappings"]["Settings"]["ActionMemory"]
memory_options = self.action_properties.get(lambda_size, [])
if not self.use_ecs and actions.ACTION_USE_ECS in memory_options:
del memory_options[memory_options.index(actions.ACTION_USE_ECS)]
if len(memory_options) > 1:
self._template_parameters[lambda_size_param] = {
"Type": "String",
"AllowedValues": memory_options,
"Default": memory_options[0],
"Description": description.format(", ".join(
"{} {}{} {}".format(m,
"(" if memory_settings[m] != "" else "", memory_settings[m],
"MB)" if memory_settings[m] != "" else "") for
m in memory_options))
}
self._parameter_labels[lambda_size_param] = {
"default": label
}
size_group["Parameters"].append(lambda_size_param)
if actions.ACTION_USE_ECS in memory_options:
self._template_parameters[config_ecs_memory_param] = {
"Type": "Number",
"MinValue": 8,
"Default": 128,
"Description": ecs_description
}
self._parameter_labels[config_ecs_memory_param] = {
"default": ecs_memory_label
}
size_group["Parameters"].append(config_ecs_memory_param)
def setup_memory_parameters():
memory_group = {
"Label": {
"default": "Task memory allocation settings",
},
"Parameters": []
}
build_memory_parameter(size_group=memory_group,
lambda_size=actions.ACTION_SELECT_SIZE,
lambda_size_param=configuration.CONFIG_TASK_SELECT_SIZE,
config_ecs_memory_param=configuration.CONFIG_ECS_SELECT_MEMORY,
description=PARAM_DESCRIPTION_SELECT_SIZE,
label=PARAM_LABEL_SELECT_MEMORY,
ecs_memory_label=PARAM_LABEL_ECS_SELECT_MEMORY,
ecs_description=PARAM_DESCRIPTION_ECS_SELECT_MEMORY)
build_memory_parameter(size_group=memory_group,
lambda_size=actions.ACTION_EXECUTE_SIZE,
lambda_size_param=configuration.CONFIG_TASK_EXECUTE_SIZE,
config_ecs_memory_param=configuration.CONFIG_ECS_EXECUTE_MEMORY,
description=PARAM_DESCRIPTION_EXECUTE_SIZE,
label=PARAM_LABEL_EXECUTION_MEMORY,
ecs_memory_label=PARAM_LABEL_ECS_EXEC_MEMORY,
ecs_description=PARAM_DESCRIPTION_ECS_EXECUTE_MEMORY)
if self.has_completion_logic:
build_memory_parameter(size_group=memory_group,
lambda_size=actions.ACTION_COMPLETION_SIZE,
lambda_size_param=configuration.CONFIG_TASK_COMPLETION_SIZE,
config_ecs_memory_param=configuration.CONFIG_ECS_COMPLETION_MEMORY,
description=PARAM_DESCRIPTION_COMPLETION_SIZE,
label=PARAM_LABEL_COMPLETION_MEMORY,
ecs_memory_label=PARAM_LABEL_COMPLETION_MEMORY,
ecs_description=PARAM_DESCRIPTION_ECS_COMPLETION_MEMORY)
if len(memory_group["Parameters"]) > 0:
self._parameter_groups.append(memory_group)
self._template_parameters.update(
{
# task enabled parameter
configuration.CONFIG_ENABLED: {
"Type": "String",
"Default": YES,
"AllowedValues": YES_NO,
"Description": PARAM_DESCRIPTION_ENABLE
},
# task debug switch parameter
configuration.CONFIG_DEBUG: {
"Type": "String",
"Default": NO,
"AllowedValues": YES_NO,
"Description": PARAM_DESCRIPTION_DEBUG
},
# task debug switch parameter
configuration.CONFIG_TASK_NOTIFICATIONS: {
"Type": "String",
"Default": NO,
"AllowedValues": YES_NO,
"Description": PARAM_DESCRIPTION_NOTIFICATIONS
},
configuration.CONFIG_TASK_METRICS: {
"Type": "String",
"Default": NO,
"AllowedValues": YES_NO,
"Description": PARAM_DESCRIPTION_METRICS
},
configuration.CONFIG_DESCRIPTION: {
"Type": "String",
"Description": PARAM_DESCRIPTION_TASK_DESCRIPTION
}
})
if self.use_intervals:
desc = PARAM_DESCRIPTION_INTERVAL if self.interval_min == 0 else PARAM_DESC_INTERVAL_MIN.format(
PARAM_DESCRIPTION_INTERVAL, self.interval_min)
self._template_parameters.update({
# task interval
configuration.CONFIG_INTERVAL: {
"Type": "String",
"Description": desc
},
# task timezone
configuration.CONFIG_TIMEZONE: {
"Type": "String",
"Default": "UTC",
"AllowedValues": pytz.all_timezones,
"Description": PARAM_DESCRIPTION_TIMEZONE
}
})
self._parameter_labels.update({
configuration.CONFIG_INTERVAL: {"default": PARAM_LABEL_INTERVAL},
configuration.CONFIG_TIMEZONE: {"default": PARAM_LABEL_TIMEZONE}
})
self._parameter_labels.update(
{
# parameter labels
configuration.CONFIG_ENABLED: {"default": PARAM_LABEL_ENABLED},
configuration.CONFIG_DEBUG: {"default": PARAM_LABEL_DEBUG},
configuration.CONFIG_TASK_METRICS: {"default": PARAM_LABEL_METRICS},
configuration.CONFIG_DESCRIPTION: {"default": PARAM_LABEL_TASK_DESCRIPTION}
})
self._parameter_groups.append(
{
# parameter groups
"Label": {
"default": GROUP_LABEL_TASK_SETTINGS.format(self.action_properties[actions.ACTION_TITLE],
self.action_properties[actions.ACTION_VERSION],
os.getenv(handlers.ENV_STACK_NAME, "")),
},
"Parameters": [
configuration.CONFIG_DESCRIPTION,
configuration.CONFIG_INTERVAL,
configuration.CONFIG_TIMEZONE,
configuration.CONFIG_ENABLED,
configuration.CONFIG_TASK_METRICS,
configuration.CONFIG_TASK_NOTIFICATIONS,
configuration.CONFIG_DEBUG
] if self.use_intervals else
[
configuration.CONFIG_DESCRIPTION,
configuration.CONFIG_ENABLED,
configuration.CONFIG_TASK_METRICS,
configuration.CONFIG_TASK_NOTIFICATIONS,
configuration.CONFIG_DEBUG
]
})
def setup_event_parameters():
self._action_events = {}
self._action_scopes = {}
if not self.use_events:
return
action_events = self.action_properties.get(actions.ACTION_EVENTS, {})
action_event_scopes = self.action_properties.get(actions.ACTION_EVENT_SCOPES, {})
# interval is required if there no events
if len(action_events) == 0:
interval_parameter = self._template["Parameters"][configuration.CONFIG_INTERVAL]
interval_parameter["MinLength"] = 9
for e in EVENTS:
event_detail_type = e[0]
handled_events = e[1]
source = handled_events[EVENT_SOURCE]
action_handled_details_types = action_events.get(source, {})
action_handled_events = action_handled_details_types.get(event_detail_type, [])
event_detail_scopes = action_event_scopes.get(source, {}).get(event_detail_type, {})
if len(action_handled_events) > 0:
parameter_group = {
"Label": {
"default": handled_events["title"]
},
"Parameters": []
}
for event_name in sorted(action_handled_events):
# if not event_name in handled_events:
# continue
event_cloudformation_parameter = {
"Type": "String",
"Default": NO,
"AllowedValues": YES_NO,
"Description": handled_events["events"][event_name][EVENT_DESCRIPTION]
}
param_name = handled_events[EVENT_PARAMETER].format(self._format_event_name(event_name))
self._template_parameters[param_name] = event_cloudformation_parameter
parameter_group["Parameters"].append(param_name)
event_param_label = handled_events[EVENT_EVENTS][event_name][EVENT_LABEL]
self._parameter_labels[param_name] = {
"default": event_param_label
}
if source not in self._action_events:
self._action_events[source] = {}
if event_detail_type not in self._action_events[source]:
self._action_events[source][event_detail_type] = []
self._action_events[source][event_detail_type].append(event_name)
event_scope = event_detail_scopes.get(event_name, handlers.EVENT_SCOPE_RESOURCE)
if event_scope == handlers.EVENT_SCOPE_REGION:
event_source_cloudformation_parameter = {
"Type": "String",
"Default": handlers.EVENT_SCOPE_RESOURCE,
"AllowedValues": [handlers.EVENT_SCOPE_RESOURCE,
handlers.EVENT_SCOPE_REGION],
"Description": PARAM_DESCRIPTION_EVENT_SCOPE.format(event_name)
}
param_name = handled_events[EVENT_SCOPE_PARAMETER].format(self._format_event_name(event_name))
self._template_parameters[param_name] = event_source_cloudformation_parameter
parameter_group["Parameters"].append(param_name)
self._parameter_labels[param_name] = {
"default": PARAM_LABEL_SCOPE.format(event_param_label.lower())
}
if source not in self._action_scopes:
self._action_scopes[source] = {}
if event_detail_type not in self._action_scopes[source]:
self._action_scopes[source][event_detail_type] = []
self._action_scopes[source][event_detail_type].append(event_name)
if len(parameter_group["Parameters"]) > 0:
self._parameter_groups.append(parameter_group)
# actions that can react to events with a wider scope than the resource that cause the event can filter on
# the tags of the source resource that cause the event
if self.action_properties.get(actions.ACTION_PARAM_EVENT_SOURCE_TAG_FILTER, False):
self._template_parameters[configuration.CONFIG_EVENT_SOURCE_TAG_FILTER] = {
"Type": "String",
"Description": PARAM_DESCRIPTION_EVENT_SOURCE_RESOURCE_TAG_FILTER.format(
self.action_properties[actions.ACTION_SERVICE].upper())
}
self._parameter_groups.append(
{
"Label": {
"default": PARAM_GROUP_EVENT_FILTERING
},
"Parameters": [configuration.CONFIG_EVENT_SOURCE_TAG_FILTER]
})
self._parameter_labels[configuration.CONFIG_EVENT_SOURCE_TAG_FILTER] = {
"default": PARAM_LABEL_EVENT_SOURCE_RESOURCE_TAG_FILTER
}
setup_region_parameter()
setup_cross_account_parameters()
setup_tag_filter_parameter()
setup_timeout_parameter()
setup_event_parameters()
setup_memory_parameters()