"""State module for managing Amazon anomaly detection monitor."""
import copy
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict
from typing import List
__contracts__ = ["resource"]
[docs]async def present(
hub,
ctx,
name: str,
monitor_name: str,
monitor_type: str,
monitor_dimension: str = None,
monitor_specification: make_dataclass(
"Expression",
[
(
"Or",
List[
make_dataclass(
"Expression",
[
("Or", List["Expression"], field(default=None)),
("And", List["Expression"], field(default=None)),
("Not", "Expression", field(default=None)),
(
"Dimensions",
make_dataclass(
"DimensionValues",
[
("Key", str, field(default=None)),
("Values", List[str], field(default=None)),
(
"MatchOptions",
List[str],
field(default=None),
),
],
),
field(default=None),
),
(
"Tags",
make_dataclass(
"TagValues",
[
("Key", str, field(default=None)),
("Values", List[str], field(default=None)),
(
"MatchOptions",
List[str],
field(default=None),
),
],
),
field(default=None),
),
(
"CostCategories",
make_dataclass(
"CostCategoryValues",
[
("Key", str, field(default=None)),
("Values", List[str], field(default=None)),
(
"MatchOptions",
List[str],
field(default=None),
),
],
),
field(default=None),
),
],
)
],
field(default=None),
),
("And", List["Expression"], field(default=None)),
("Not", "Expression", field(default=None)),
("Dimensions", "DimensionValues", field(default=None)),
("Tags", Any, field(default=None)),
("CostCategories", Any, field(default=None)),
],
) = None,
resource_id: str = None,
) -> Dict[str, Any]:
"""Creates a new cost anomaly detection monitor with the requested type and monitor specification.
Args:
name(str):
An Idem name of the resource.
monitor_name(str):
Name of anomaly monitor.
monitor_type(str):
Possible type values. Valid Values: DIMENSIONAL | CUSTOM
monitor_dimension(str):
The dimensions to evaluate : SERVICE
resource_id(str, Optional):
Monitor ARN to identify the resource.
monitor_specification(dict[str, Any], Optional):
Use Expression to filter by cost or by usage. There are two patterns:
* Simple dimension values
You can set the dimension name and values for the filters that you plan to use. For example,
you can filter for REGION==us-east-1 OR REGION==us-west-1. For GetRightsizingRecommendation, the
Region is a full name (for example, REGION==US East (N. Virginia). The Expression example is as
follows: ``{ "Dimensions": { "Key": "REGION", "Values": [ "us-east-1", “us-west-1” ] } }`` The
list of dimension values are OR'd together to retrieve cost or usage data. You can create
Expression and DimensionValues objects using either with* methods or set* methods in multiple
lines.
* Compound dimension values with logical operations
You can use multiple Expression types and the logical operators AND/OR/NOT to create a list of one or more Expression objects.
By doing this, you can filter on more advanced options. For example, you can filter on ((REGION
== us-east-1 OR REGION == us-west-1) OR (TAG.Type == Type1)) AND (USAGE_TYPE != DataTransfer).
The Expression for that is as follows: ``{ "And": [ {"Or": [ {"Dimensions": { "Key": "REGION",
"Values": [ "us-east-1", "us-west-1" ] }}, {"Tags": { "Key": "TagName", "Values": ["Value1"] } }
]}, {"Not": {"Dimensions": { "Key": "USAGE_TYPE", "Values": ["DataTransfer"] }}} ] }``
.. Note::
Because each Expression can have only one operator, the service returns an error if more than one is
specified. The following example shows an Expression object that creates an error. ``{ "And": [
... ], "DimensionValues": { "Dimension": "USAGE_TYPE", "Values": [ "DataTransfer" ] } }``
.. Note::
For the GetRightsizingRecommendation action, a combination of OR and NOT isn't supported. OR isn't
supported between different dimensions, or dimensions and tags. NOT operators aren't supported.
Dimensions are also limited to LINKED_ACCOUNT, REGION, or RIGHTSIZING_TYPE. For the
GetReservationPurchaseRecommendation action, only NOT is supported. AND and OR aren't supported.
Dimensions are limited to LINKED_ACCOUNT.
* Or (list[Expression], Optional):
Return results that match either Dimension object.
* Or (list[Expression], Optional):
Return results that match either Dimension object.
* And (list[Expression], Optional):
Return results that match both Dimension objects.
* Not (Expression, Optional):
Return results that don't match a Dimension object.
* Dimensions (DimensionValues, Optional):
The specific Dimension to use for Expression.
* Key (str, Optional):
The names of the metadata types that you can use to filter and group your results. For example,
AZ returns a list of Availability Zones.
* Values (list[str], Optional):
The metadata values that you can use to filter and group your results. You can use
GetDimensionValues to find specific values.
* MatchOptions (List[str], Optional):
The match options that you can use to filter your results. MatchOptions is only applicable for
actions related to Cost Category. The default values for MatchOptions are EQUALS and
CASE_SENSITIVE.
* Tags (TagValues, Optional):
The specific Tag to use for Expression.
* Key (str, Optional):
The key for the tag.
* Values (list[str], Optional):
The specific value of the tag.
* MatchOptions (list[str], Optional):
The match options that you can use to filter your results. MatchOptions is only applicable for
actions related to Cost Category. The default values for MatchOptions are EQUALS and
CASE_SENSITIVE.
* CostCategories (CostCategoryValues, Optional):
The filter that's based on CostCategory values.
* Key (str, Optional):
The unique name of the Cost Category.
* Values (list[str], Optional):
The specific value of the Cost Category.
* MatchOptions (list[str], Optional):
The match options that you can use to filter your results. MatchOptions is only applicable for
actions related to cost category. The default values for MatchOptions is EQUALS and
CASE_SENSITIVE.
* And (list[Expression], Optional):
Return results that match both Dimension objects.
* Not (Expression, Optional):
Return results that don't match a Dimension object.
* Dimensions (DimensionValues, Optional):
The specific Dimension to use for Expression.
* Tags (Any, Optional):
The specific Tag to use for Expression.
* CostCategories (Any, Optional):
The filter that's based on CostCategory values.
Request Syntax:
.. code-block:: sls
[monitor-resource-id]:
aws.costexplorer.anomaly_monitor.present:
- resource_id: "string"
- monitor_name: "string"
- monitor_type: "string"
- monitor_specification:
- Dimensions: "dict"
Returns:
Dict[str, str]
Examples:
.. code-block:: sls
cost_monitor1234:
aws.costexplorer.anomaly_monitor.present:
- name: cost_monitor1234
- resource_id: cost_monitor1234
- monitor_specification:
Dimensions:
Key: LINKED_ACCOUNT
Values:
- "820272282974"
- monitor_type: CUSTOM
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
before = None
resource_updated = False
if resource_id:
before = await hub.exec.boto3.client.ce.get_anomaly_monitors(
ctx, MonitorArnList=[resource_id]
)
if before and before["ret"].get("AnomalyMonitors"):
before = before["ret"]["AnomalyMonitors"][0]
else:
before = None
if not before:
if ctx.get("test", False):
result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={},
desired_state={
"name": name,
"anomaly_monitor": {
"monitor_name": monitor_name,
"monitor_specification": monitor_specification,
"monitor_type": monitor_type,
"monitor_dimension": monitor_dimension,
},
},
)
result["comment"] += [
f"Would create aws.costexplorer.anomaly_monitor {name}"
]
return result
monitor = None
if monitor_specification:
monitor = {
"MonitorName": monitor_name,
"MonitorSpecification": monitor_specification,
"MonitorType": monitor_type,
}
else:
monitor = {
"MonitorName": monitor_name,
"MonitorDimension": monitor_dimension,
"MonitorType": monitor_type,
}
ret = await hub.exec.boto3.client.ce.create_anomaly_monitor(
ctx,
AnomalyMonitor=monitor,
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] = ret["comment"]
return result
result["comment"] += [f"Created aws.costexplorer.anomaly_monitor '{name}'"]
resource_id = ret["ret"]["MonitorArn"]
else:
result[
"old_state"
] = hub.tool.aws.costexplorer.conversion_utils.convert_raw_monitor_to_present(
ctx, raw_resource=before, idem_resource_name=name
)
plan_state = copy.deepcopy(result["old_state"])
resource_id = before.get("MonitorArn")
update_ret = await hub.tool.aws.costexplorer.anomaly_monitor.update_monitor(
ctx, before=before, monitor_name=monitor_name, monitor_arn=resource_id
)
result["comment"] += update_ret["comment"]
result["result"] = update_ret["result"]
resource_updated = resource_updated or bool(update_ret["ret"])
if update_ret["ret"] and ctx.get("test", False):
if "monitor_name" in update_ret["ret"]:
plan_state["monitor_name"] = update_ret["ret"]["monitor_name"]
if not resource_updated:
result["comment"] += [f"{name} already exists"]
try:
if ctx.get("test", False):
result["new_state"] = plan_state
elif (not before) or resource_updated:
after = await hub.exec.boto3.client.ce.get_anomaly_monitors(
ctx, MonitorArnList=[resource_id]
)
result[
"new_state"
] = hub.tool.aws.costexplorer.conversion_utils.convert_raw_monitor_to_present(
ctx,
raw_resource=after["ret"]["AnomalyMonitors"][0],
idem_resource_name=name,
)
else:
result["new_state"] = copy.deepcopy(result["old_state"])
except Exception as e:
result["comment"] += [str(e)]
result["result"] = False
return result
[docs]async def absent(hub, ctx, name: str, resource_id: str = None) -> Dict[str, Any]:
"""Deletes a cost anomaly monitor by the specified monitor ARN as resource_id.
Args:
name(str):
The Idem name of the anomaly monitor.
resource_id(str, Optional):
Monitor ARN to identify the resource.
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
cost_monitor1234:
aws.costexplorer.anomaly_monitor.absent:
- name: value
- resource_id: cost_monitor1234
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
if not resource_id:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.costexplorer.anomaly_monitor", name=name
)
return result
before = await hub.exec.boto3.client.ce.get_anomaly_monitors(
ctx, MonitorArnList=[resource_id]
)
if before and before["ret"].get("AnomalyMonitors"):
before = before["ret"]["AnomalyMonitors"][0]
else:
before = None
if not before:
result["comment"] += [
f"aws.costexplorer.anomaly_monitor '{name}' already absent"
]
elif ctx.get("test", False):
result[
"old_state"
] = hub.tool.aws.costexplorer.conversion_utils.convert_raw_monitor_to_present(
ctx, raw_resource=before, idem_resource_name=name
)
result["comment"] += [f"Would delete aws.costexplorer.anomaly_monitor {name}"]
return result
else:
result[
"old_state"
] = hub.tool.aws.costexplorer.conversion_utils.convert_raw_monitor_to_present(
ctx, raw_resource=before, idem_resource_name=name
)
ret = await hub.exec.boto3.client.ce.delete_anomaly_monitor(
ctx, MonitorArn=resource_id
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] = ret["comment"]
return result
result["comment"] += [f"Deleted aws.costexplorer.anomaly_monitor '{name}'"]
return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
"""Returns a list of aws cost anomaly monitors.
Describe the resource in a way that can be recreated/managed with the corresponding "present" function.
If a monitor arn or name is specified, the list contains only the description of that monitor.
Returns:
Dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe aws.costexplorer.anomaly_monitor
"""
result = {}
ret = await hub.exec.boto3.client.ce.get_anomaly_monitors(ctx)
if not ret["result"]:
hub.log.warning(f"Could not describe cost anomaly monitors {ret['comment']}")
return {}
for monitor in ret["ret"]["AnomalyMonitors"]:
resource_id = monitor.get("MonitorArn")
resource_translated = (
hub.tool.aws.costexplorer.conversion_utils.convert_raw_monitor_to_present(
ctx, raw_resource=monitor, idem_resource_name=resource_id
)
)
result[resource_id] = {
"aws.costexplorer.anomaly_monitor.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in resource_translated.items()
]
}
return result