# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 John Mille <john@compose-x.io>
"""
Main module to create x-alarms defined at the top level.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from ecs_composex.alarms.alarms_helpers import create_alarms
if TYPE_CHECKING:
from ecs_composex.common.settings import ComposeXSettings
from ecs_composex.mods_manager import XResourceModule
import warnings
from compose_x_common.compose_x_common import keyisset
from troposphere import GetAtt, Ref
from troposphere.cloudwatch import CompositeAlarm
from ecs_composex.alarms.alarms_elbv2 import (
handle_elbv2_dimension_mapping,
handle_elbv2_target_group_dimensions,
)
from ecs_composex.alarms.alarms_params import ALARM_ARN, ALARM_NAME
from ecs_composex.common.logging import LOG
from ecs_composex.common.stacks import ComposeXStack
from ecs_composex.common.troposphere_tools import build_template
from ecs_composex.compose.x_resources.services_resources import ServicesXResource
[docs]class Alarm(ServicesXResource):
"""
Class to represent CW Alarms
"""
topics_key = "Topics"
def __init__(
self, name, definition, module: XResourceModule, settings: ComposeXSettings
):
self.topics = []
self.is_composite = False
self.in_composite = False
super().__init__(name, definition, module, settings)
self.topics = (
definition[self.topics_key]
if keyisset(self.topics_key, self.definition)
else []
)
[docs] def init_outputs(self):
self.output_properties = {
ALARM_NAME: (self.logical_name, self.cfn_resource, Ref, None),
ALARM_ARN: (
f"{self.logical_name}{ALARM_ARN.return_value}",
self.cfn_resource,
GetAtt,
ALARM_ARN.return_value,
),
}
[docs] def handle_dimensions(self, settings: ComposeXSettings) -> None:
"""Handles the dimensions settings and tries to resolve"""
if not hasattr(self.cfn_resource, "Dimensions"):
LOG.debug(f"{self.module.res_key}.{self.name} - No Dimensions defined")
return
dimensions = getattr(self.cfn_resource, "Dimensions")
namespace = self.cfn_resource.Namespace
if namespace == "AWS/ApplicationELB" or namespace == "AWS/NetworkELB":
for dimension in dimensions:
if dimension.Name == "LoadBalancer" and dimension.Value.startswith(
r"x-elbv2::"
):
handle_elbv2_dimension_mapping(
self.stack, dimension, self, settings
)
elif dimension.Name == "TargetGroup" and dimension.Value.startswith(
r"x-elbv2::"
):
handle_elbv2_target_group_dimensions(
self.stack, dimension, self, settings
)
[docs] def handle_x_dependencies(
self, settings: ComposeXSettings, root_stack: ComposeXStack = None
) -> None:
"""
Function to cross reference alarm settings with other resources
:param ecs_composex.common.stacks.ComposeXStacks root_stack:
:param ecs_composex.common.settings.ComposeXSettings settings:
"""
if (
isinstance(self, Alarm)
and isinstance(self.cfn_resource, CompositeAlarm)
or not self.cfn_resource
):
return
self.handle_dimensions(settings)
[docs]class XStack(ComposeXStack):
"""
Class to represent the root stack for alarms
"""
def __init__(
self, name, settings: ComposeXSettings, module: XResourceModule, **kwargs
):
if module.new_resources:
template = build_template(
f"Alarms created for {settings.name} by Compose-X"
)
super().__init__(name, stack_template=template, **kwargs)
create_alarms(template, self, module.new_resources, settings)
if not hasattr(self, "DeletionPolicy"):
setattr(self, "DeletionPolicy", module.module_deletion_policy)
else:
self.is_void = True
if module.lookup_resources:
warnings.warn(
f"{module.res_key} - Lookup and Use are not supported. "
"You can only create new resources"
)
for resource in module.resources_list:
resource.stack = self