import json
from moto.core.utils import amzn_request_id
from moto.core.responses import BaseResponse
from .models import cloudwatch_backends
from dateutil.parser import parse as dtparse
class CloudWatchResponse(BaseResponse):
@property
def cloudwatch_backend(self):
return cloudwatch_backends[self.region]
def _error(self, code, message, status=400):
template = self.response_template(ERROR_RESPONSE_TEMPLATE)
return template.render(code=code, message=message), dict(status=status)
@amzn_request_id
def put_metric_alarm(self):
name = self._get_param('AlarmName')
namespace = self._get_param('Namespace')
metric_name = self._get_param('MetricName')
comparison_operator = self._get_param('ComparisonOperator')
evaluation_periods = self._get_param('EvaluationPeriods')
period = self._get_param('Period')
threshold = self._get_param('Threshold')
statistic = self._get_param('Statistic')
description = self._get_param('AlarmDescription')
dimensions = self._get_list_prefix('Dimensions.member')
alarm_actions = self._get_multi_param('AlarmActions.member')
ok_actions = self._get_multi_param('OKActions.member')
insufficient_data_actions = self._get_multi_param(
"InsufficientDataActions.member")
unit = self._get_param('Unit')
alarm = self.cloudwatch_backend.put_metric_alarm(name, namespace, metric_name,
comparison_operator,
evaluation_periods, period,
threshold, statistic,
description, dimensions,
alarm_actions, ok_actions,
insufficient_data_actions,
unit)
template = self.response_template(PUT_METRIC_ALARM_TEMPLATE)
return template.render(alarm=alarm)
@amzn_request_id
def describe_alarms(self):
action_prefix = self._get_param('ActionPrefix')
alarm_name_prefix = self._get_param('AlarmNamePrefix')
alarm_names = self._get_multi_param('AlarmNames.member')
state_value = self._get_param('StateValue')
if action_prefix:
alarms = self.cloudwatch_backend.get_alarms_by_action_prefix(
action_prefix)
elif alarm_name_prefix:
alarms = self.cloudwatch_backend.get_alarms_by_alarm_name_prefix(
alarm_name_prefix)
elif alarm_names:
alarms = self.cloudwatch_backend.get_alarms_by_alarm_names(alarm_names)
elif state_value:
alarms = self.cloudwatch_backend.get_alarms_by_state_value(state_value)
else:
alarms = self.cloudwatch_backend.get_all_alarms()
template = self.response_template(DESCRIBE_ALARMS_TEMPLATE)
return template.render(alarms=alarms)
@amzn_request_id
def delete_alarms(self):
alarm_names = self._get_multi_param('AlarmNames.member')
self.cloudwatch_backend.delete_alarms(alarm_names)
template = self.response_template(DELETE_METRIC_ALARMS_TEMPLATE)
return template.render()
@amzn_request_id
def put_metric_data(self):
namespace = self._get_param('Namespace')
metric_data = self._get_multi_param('MetricData.member')
self.cloudwatch_backend.put_metric_data(namespace, metric_data)
template = self.response_template(PUT_METRIC_DATA_TEMPLATE)
return template.render()
@amzn_request_id
def get_metric_statistics(self):
namespace = self._get_param('Namespace')
metric_name = self._get_param('MetricName')
start_time = dtparse(self._get_param('StartTime'))
end_time = dtparse(self._get_param('EndTime'))
period = int(self._get_param('Period'))
statistics = self._get_multi_param("Statistics.member")
# Unsupported Parameters (To Be Implemented)
unit = self._get_param('Unit')
extended_statistics = self._get_param('ExtendedStatistics')
dimensions = self._get_param('Dimensions')
if unit or extended_statistics or dimensions:
raise NotImplemented()
# TODO: this should instead throw InvalidParameterCombination
if not statistics:
raise NotImplemented("Must specify either Statistics or ExtendedStatistics")
datapoints = self.cloudwatch_backend.get_metric_statistics(namespace, metric_name, start_time, end_time, period, statistics)
template = self.response_template(GET_METRIC_STATISTICS_TEMPLATE)
return template.render(label=metric_name, datapoints=datapoints)
@amzn_request_id
def list_metrics(self):
metrics = self.cloudwatch_backend.get_all_metrics()
template = self.response_template(LIST_METRICS_TEMPLATE)
return template.render(metrics=metrics)
@amzn_request_id
def delete_dashboards(self):
dashboards = self._get_multi_param('DashboardNames.member')
if dashboards is None:
return self._error('InvalidParameterValue', 'Need at least 1 dashboard')
status, error = self.cloudwatch_backend.delete_dashboards(dashboards)
if not status:
return self._error('ResourceNotFound', error)
template = self.response_template(DELETE_DASHBOARD_TEMPLATE)
return template.render()
@amzn_request_id
def describe_alarm_history(self):
raise NotImplementedError()
@amzn_request_id
def describe_alarms_for_metric(self):
raise NotImplementedError()
@amzn_request_id
def disable_alarm_actions(self):
raise NotImplementedError()
@amzn_request_id
def enable_alarm_actions(self):
raise NotImplementedError()
@amzn_request_id
def get_dashboard(self):
dashboard_name = self._get_param('DashboardName')
dashboard = self.cloudwatch_backend.get_dashboard(dashboard_name)
if dashboard is None:
return self._error('ResourceNotFound', 'Dashboard does not exist')
template = self.response_template(GET_DASHBOARD_TEMPLATE)
return template.render(dashboard=dashboard)
@amzn_request_id
def list_dashboards(self):
prefix = self._get_param('DashboardNamePrefix', '')
dashboards = self.cloudwatch_backend.list_dashboards(prefix)
template = self.response_template(LIST_DASHBOARD_RESPONSE)
return template.render(dashboards=dashboards)
@amzn_request_id
def put_dashboard(self):
name = self._get_param('DashboardName')
body = self._get_param('DashboardBody')
try:
json.loads(body)
except ValueError:
return self._error('InvalidParameterInput', 'Body is invalid JSON')
self.cloudwatch_backend.put_dashboard(name, body)
template = self.response_template(PUT_DASHBOARD_RESPONSE)
return template.render()
@amzn_request_id
def set_alarm_state(self):
alarm_name = self._get_param('AlarmName')
reason = self._get_param('StateReason')
reason_data = self._get_param('StateReasonData')
state_value = self._get_param('StateValue')
self.cloudwatch_backend.set_alarm_state(alarm_name, reason, reason_data, state_value)
template = self.response_template(SET_ALARM_STATE_TEMPLATE)
return template.render()
PUT_METRIC_ALARM_TEMPLATE = """
{{ request_id }}
"""
DESCRIBE_ALARMS_TEMPLATE = """
{% for alarm in alarms %}
{{ alarm.actions_enabled }}
{% for action in alarm.alarm_actions %}
{{ action }}
{% endfor %}
{{ alarm.arn }}
{{ alarm.configuration_updated_timestamp }}
{{ alarm.description }}
{{ alarm.name }}
{{ alarm.comparison_operator }}
{% for dimension in alarm.dimensions %}
{{ dimension.name }}
{{ dimension.value }}
{% endfor %}
{{ alarm.evaluation_periods }}
{% for action in alarm.insufficient_data_actions %}
{{ action }}
{% endfor %}
{{ alarm.metric_name }}
{{ alarm.namespace }}
{% for action in alarm.ok_actions %}
{{ action }}
{% endfor %}
{{ alarm.period }}
{{ alarm.state_reason }}
{{ alarm.state_reason_data }}
{{ alarm.state_updated_timestamp }}
{{ alarm.state_value }}
{{ alarm.statistic }}
{{ alarm.threshold }}
{{ alarm.unit }}
{% endfor %}
"""
DELETE_METRIC_ALARMS_TEMPLATE = """
{{ request_id }}
"""
PUT_METRIC_DATA_TEMPLATE = """
{{ request_id }}
"""
GET_METRIC_STATISTICS_TEMPLATE = """
{{ request_id }}
{% for datapoint in datapoints %}
{% if datapoint.sum %}
{{ datapoint.sum }}
{% endif %}
{% if datapoint.average %}
{{ datapoint.average }}
{% endif %}
{% if datapoint.maximum %}
{{ datapoint.maximum }}
{% endif %}
{% if datapoint.minimum %}
{{ datapoint.minimum }}
{% endif %}
{% if datapoint.sample_count %}
{{ datapoint.sample_count }}
{% endif %}
{% if datapoint.extended_statistics %}
{{ datapoint.extended_statistics }}
{% endif %}
{{ datapoint.timestamp }}
{{ datapoint.unit }}
{% endfor %}
"""
LIST_METRICS_TEMPLATE = """
{% for metric in metrics %}
{% for dimension in metric.dimensions %}
{{ dimension.name }}
{{ dimension.value }}
{% endfor %}
{{ metric.name }}
{{ metric.namespace }}
{% endfor %}
96e88479-4662-450b-8a13-239ded6ce9fe
"""
PUT_DASHBOARD_RESPONSE = """
{{ request_id }}
"""
LIST_DASHBOARD_RESPONSE = """
{% for dashboard in dashboards %}
{{ dashboard.arn }}
{{ dashboard.last_modified_iso }}
{{ dashboard.size }}
{{ dashboard.name }}
{% endfor %}
{{ request_id }}
"""
DELETE_DASHBOARD_TEMPLATE = """
{{ request_id }}
"""
GET_DASHBOARD_TEMPLATE = """
{{ dashboard.arn }}
{{ dashboard.body }}
{{ dashboard.name }}
{{ request_id }}
"""
SET_ALARM_STATE_TEMPLATE = """
{{ request_id }}
"""
ERROR_RESPONSE_TEMPLATE = """
Sender
{{ code }}
{{ message }}
{{ request_id }}
"""