ApplicationAutoscaling: put_scaling_policy() now generates CW alarms …

…(for some service-types) (#7566)

moto/applicationautoscaling/models.py

@@ -1,7 +1,7 @@
 import time
 from collections import OrderedDict
 from enum import Enum, unique
-from typing import Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -10,6 +10,9 @@
 
 from .exceptions import AWSValidationException
 
+if TYPE_CHECKING:
+    from moto.cloudwatch.models import FakeAlarm
+
 
 @unique
 class ResourceTypeExceptionValueSet(Enum):
@@ -76,7 +79,6 @@ def __init__(self, region_name: str, account_id: str) -> None:
     def describe_scalable_targets(
         self, namespace: str, r_ids: Union[None, List[str]], dimension: Union[None, str]
     ) -> List["FakeScalableTarget"]:
-        """Describe scalable targets."""
         if r_ids is None:
             r_ids = []
         targets = self._flatten_scalable_targets(namespace)
@@ -105,7 +107,6 @@ def register_scalable_target(
         role_arn: str,
         suspended_state: str,
     ) -> "FakeScalableTarget":
-        """Registers or updates a scalable target."""
         _ = _target_params_are_valid(namespace, r_id, dimension)
         if namespace == ServiceNamespaceValueSet.ECS.value:
             _ = self._ecs_service_exists_for_target(r_id)
@@ -151,7 +152,6 @@ def _add_scalable_target(
     def deregister_scalable_target(
         self, namespace: str, r_id: str, dimension: str
     ) -> None:
-        """Registers or updates a scalable target."""
         if self._scalable_target_exists(r_id, dimension):
             del self.targets[dimension][r_id]
         else:
@@ -165,7 +165,7 @@ def put_scaling_policy(
         service_namespace: str,
         resource_id: str,
         scalable_dimension: str,
-        policy_body: str,
+        policy_body: Dict[str, Any],
         policy_type: Optional[None],
     ) -> "FakeApplicationAutoscalingPolicy":
         policy_key = FakeApplicationAutoscalingPolicy.formulate_key(
@@ -238,6 +238,8 @@ def delete_scaling_policy(
             service_namespace, resource_id, scalable_dimension, policy_name
         )
         if policy_key in self.policies:
+            policy = self.policies[policy_key]
+            policy.delete_alarms(self.account_id, self.region_name)
             del self.policies[policy_key]
         else:
             raise AWSValidationException(
@@ -438,7 +440,7 @@ def __init__(
         resource_id: str,
         scalable_dimension: str,
         policy_type: Optional[str],
-        policy_body: str,
+        policy_body: Dict[str, Any],
     ) -> None:
         self.step_scaling_policy_configuration = None
         self.target_tracking_scaling_policy_configuration = None
@@ -451,7 +453,7 @@ def __init__(
             self.target_tracking_scaling_policy_configuration = policy_body
         else:
             raise AWSValidationException(
-                f"Unknown policy type {policy_type} specified."
+                f"1 validation error detected: Value '{policy_type}' at 'policyType' failed to satisfy constraint: Member must satisfy enum value set: [PredictiveScaling, StepScaling, TargetTrackingScaling]"
             )
 
         self._policy_body = policy_body
@@ -463,6 +465,151 @@ def __init__(
         self._guid = mock_random.uuid4()
         self.policy_arn = f"arn:aws:autoscaling:{region_name}:{account_id}:scalingPolicy:{self._guid}:resource/{self.service_namespace}/{self.resource_id}:policyName/{self.policy_name}"
         self.creation_time = time.time()
+        self.alarms: List["FakeAlarm"] = []
+
+        self.account_id = account_id
+        self.region_name = region_name
+
+        self.create_alarms()
+
+    def create_alarms(self) -> None:
+        if self.policy_type == "TargetTrackingScaling":
+            if self.service_namespace == "dynamodb":
+                self.alarms.extend(self._generate_dynamodb_alarms())
+            if self.service_namespace == "ecs":
+                self.alarms.extend(self._generate_ecs_alarms())
+
+    def _generate_dynamodb_alarms(self) -> List["FakeAlarm"]:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[self.account_id][
+            self.region_name
+        ]
+        alarms = []
+        table_name = self.resource_id.split("/")[-1]
+        alarm_action = f"{self.policy_arn}:createdBy/{mock_random.uuid4()}"
+        alarm1 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-AlarmHigh-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ConsumedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=2,
+            period=60,
+            threshold=42.0,
+            statistic="Sum",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm1)
+        alarm2 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-AlarmLow-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ConsumedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=15,
+            period=60,
+            threshold=30.0,
+            statistic="Sum",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm2)
+        alarm3 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-ProvisionedCapacityHigh-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ProvisionedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=2,
+            period=300,
+            threshold=1.0,
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm3)
+        alarm4 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-ProvisionedCapacityLow-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ProvisionedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=3,
+            period=300,
+            threshold=1.0,
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm4)
+        return alarms
+
+    def _generate_ecs_alarms(self) -> List["FakeAlarm"]:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[self.account_id][
+            self.region_name
+        ]
+        alarms: List["FakeAlarm"] = []
+        alarm_action = f"{self.policy_arn}:createdBy/{mock_random.uuid4()}"
+        config = self.target_tracking_scaling_policy_configuration or {}
+        metric_spec = config.get("PredefinedMetricSpecification", {})
+        if "Memory" in metric_spec.get("PredefinedMetricType", ""):
+            metric_name = "MemoryUtilization"
+        else:
+            metric_name = "CPUUtilization"
+        _, cluster_name, service_name = self.resource_id.split("/")
+        alarm1 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-{self.resource_id}-AlarmHigh-{mock_random.uuid4()}",
+            namespace="AWS/ECS",
+            metric_name=metric_name,
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=3,
+            period=60,
+            threshold=6,
+            unit="Percent",
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[
+                {"name": "ClusterName", "value": cluster_name},
+                {"name": "ServiceName", "value": service_name},
+            ],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm1)
+        alarm2 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-{self.resource_id}-AlarmLow-{mock_random.uuid4()}",
+            namespace="AWS/ECS",
+            metric_name=metric_name,
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=15,
+            period=60,
+            threshold=6,
+            unit="Percent",
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[
+                {"name": "ClusterName", "value": cluster_name},
+                {"name": "ServiceName", "value": service_name},
+            ],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm2)
+        return alarms
+
+    def delete_alarms(self, account_id: str, region_name: str) -> None:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[account_id][region_name]
+        cloudwatch.delete_alarms([a.name for a in self.alarms])
 
     @staticmethod
     def formulate_key(

moto/applicationautoscaling/responses.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from moto.core.responses import BaseResponse
 
@@ -79,7 +79,9 @@ def put_scaling_policy(self) -> str:
                 self._get_param("TargetTrackingScalingPolicyConfiguration"),
             ),
         )
-        return json.dumps({"PolicyARN": policy.policy_arn, "Alarms": []})  # ToDo
+        return json.dumps(
+            {"PolicyARN": policy.policy_arn, "Alarms": _build_alarms(policy)}
+        )
 
     def describe_scaling_policies(self) -> str:
         (
@@ -205,6 +207,10 @@ def _build_target(t: FakeScalableTarget) -> Dict[str, Any]:
     }
 
 
+def _build_alarms(policy: FakeApplicationAutoscalingPolicy) -> List[Dict[str, str]]:
+    return [{"AlarmARN": a.alarm_arn, "AlarmName": a.name} for a in policy.alarms]
+
+
 def _build_policy(p: FakeApplicationAutoscalingPolicy) -> Dict[str, Any]:
     response = {
         "PolicyARN": p.policy_arn,
@@ -214,6 +220,7 @@ def _build_policy(p: FakeApplicationAutoscalingPolicy) -> Dict[str, Any]:
         "ScalableDimension": p.scalable_dimension,
         "PolicyType": p.policy_type,
         "CreationTime": p.creation_time,
+        "Alarms": _build_alarms(p),
     }
     if p.policy_type == "StepScaling":
         response["StepScalingPolicyConfiguration"] = p.step_scaling_policy_configuration

tests/test_applicationautoscaling/__init__.py

@@ -0,0 +1,71 @@
+import os
+from functools import wraps
+
+import boto3
+
+from moto import mock_aws
+from tests.test_dynamodb import dynamodb_aws_verified
+
+
+def application_autoscaling_aws_verified():
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_aws` context.
+
+    This decorator will:
+      - Register a Scalable Target
+      - Run the test
+      - Deregister the Scalable Target
+    """
+
+    def inner(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            allow_aws_request = (
+                os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+            )
+
+            @dynamodb_aws_verified()
+            def scalable_target_with_dynamodb_table(table_name):
+                namespace = "dynamodb"
+                resource_id = f"table/{table_name}"
+                scalable_dimension = "dynamodb:table:ReadCapacityUnits"
+
+                client = boto3.client(
+                    "application-autoscaling", region_name="us-east-1"
+                )
+                kwargs["client"] = client
+                kwargs["namespace"] = namespace
+                kwargs["resource_id"] = resource_id
+                kwargs["scalable_dimension"] = scalable_dimension
+
+                client.register_scalable_target(
+                    ServiceNamespace=namespace,
+                    ResourceId=resource_id,
+                    ScalableDimension=scalable_dimension,
+                    MinCapacity=1,
+                    MaxCapacity=4,
+                )
+                try:
+                    resp = func(*args, **kwargs)
+                finally:
+                    client.deregister_scalable_target(
+                        ServiceNamespace=namespace,
+                        ResourceId=resource_id,
+                        ScalableDimension=scalable_dimension,
+                    )
+
+                return resp
+
+            if allow_aws_request:
+                return scalable_target_with_dynamodb_table()
+            else:
+                with mock_aws():
+                    return scalable_target_with_dynamodb_table()
+
+        return wrapper
+
+    return inner