☰
Moto 中 application-autoscaling(AWS Application Auto Scaling)服务的模拟实现指南
2026/9/25 2:28:57 网站建设 项目流程
  • Mock
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载

本指南围绕 Moto 仓库中对 AWS Application Auto Scaling 服务的模拟支持展开,系统梳理其已实现/未实现的 API 操作、可扩展目标(Scalable Target)、扩缩容策略(Scaling Policy)与定时操作(Scheduled Action)的完整用法,并结合源码与测试给出可直接运行的 boto3 示例,帮助你快速在本地测试、CI 中还原真实的 Auto Scaling 行为。

一、服务概述与实现状态总览

1.1 什么是 Application Auto Scaling

AWS Application Auto Scaling 是一项托管服务,用于自动调整可扩展资源的容量。典型场景包括:

  • ECS 服务(ecs:service:DesiredCount)的动态扩缩容;
  • DynamoDB 表的读写容量单位(dynamodb:table:ReadCapacityUnits/WriteCapacityUnits);
  • DynamoDB 全局二级索引(dynamodb:index:*);
  • SageMaker 终端节点变体实例数(sagemaker:variant:DesiredInstanceCount);
  • Lambda 预置并发(lambda:function:ProvisionedConcurrency)等。

Moto 对该服务进行模拟,使开发者可以在不触碰真实 AWS 资源的前提下,测试依赖 Auto Scaling 的应用逻辑。

1.2 官方文档中的实现清单

根据 docs/docs/services/application-autoscaling.rst 中列出的特性清单,Moto 当前实现状态如下:

操作状态
delete_scaling_policy✅ 已实现
delete_scheduled_action✅ 已实现
deregister_scalable_target✅ 已实现
describe_scalable_targets✅ 已实现
describe_scaling_activities❌ 未实现
describe_scaling_policies✅ 已实现
describe_scheduled_actions✅ 已实现
get_predictive_scaling_forecast❌ 未实现
list_tags_for_resource❌ 未实现
put_scaling_policy✅ 已实现
put_scheduled_action✅ 已实现
register_scalable_target✅ 已实现
tag_resource❌ 未实现
untag_resource❌ 未实现

文档同时明确指出:

Pagination is not yet implemented

即分页尚未实现(不过从源码看,describe_scalable_targets与describe_scaling_policies已具备 NextToken 处理逻辑,具体见下文分析)。

1.3 未实现操作的说明

  • describe_scaling_activities:无法查询扩容/缩容活动历史;
  • get_predictive_scaling_forecast:预测式扩缩容(Predictive Scaling)的预测能力未模拟;
  • tag_resource/untag_resource/list_tags_for_resource:标签管理操作暂不支持。

同时,put_scaling_policy仅支持StepScaling与TargetTrackingScaling两种策略类型(源码中FakeApplicationAutoscalingPolicy对PredictiveScaling会直接抛出ValidationException,见 moto/applicationautoscaling/models.py)。

二、源码结构与请求处理链路

2.1 模块文件组成

Moto 的 application-autoscaling 模拟模块位于 moto/applicationautoscaling/,由以下文件组成:

文件职责
models.py后端核心模型:可扩展目标、扩缩容策略、定时操作及参数校验
responses.py接收 HTTP 请求、解析参数、调用后端并序列化 JSON 响应
urls.py路由注册,匹配application-autoscaling.{region}.amazonaws.com端点
exceptions.pyAWSValidationException异常定义
utils.py从请求 URL 提取 region 的工具函数

2.2 路由与请求分发

在 urls.py 中,URL 基址为:

url_bases = [r"https?://application-autoscaling\.(.+)\.amazonaws.com"]

所有 POST 请求统一由ApplicationAutoScalingResponse.dispatch分发。响应类在 responses.py 中,通过applicationautoscaling_backends[self.current_account][self.region]获取当前账号、当前区域的独立后端实例。这意味着不同账号、不同区域之间的数据彼此隔离,与真实 AWS 的多区域行为一致。

2.3 后端数据模型

ApplicationAutoscalingBackend(models.py)内部维护三类内存数据:

self.targets: dict[str, dict[str, FakeScalableTarget]] = OrderedDict() self.policies: dict[str, FakeApplicationAutoscalingPolicy] = {} self.scheduled_actions: list[FakeScheduledAction] = []
  • targets:按scalable_dimension分组、再按resource_id索引的嵌套字典,因此同一个资源可以针对不同维度分别注册;
  • policies:以服务命名空间\t资源ID\t可扩展维度\t策略名组合键存储(formulate_key,见 models.py),确保策略定位唯一;
  • scheduled_actions:按创建顺序追加的列表。

每个FakeScalableTarget在创建时即生成形如arn:{partition}:application-autoscaling:{region}:{account_id}:scalable-target/{36位随机串}的 ARN(见 models.py),其中 partition 根据区域自动推导(如aws/aws-cn/aws-us-gov)。

三、可扩展目标(Scalable Target)的注册与管理

3.1 register_scalable_target:注册或更新目标

这是使用该服务的第一步:将某个资源注册为可扩展目标,并设定容量上下限。核心参数如下:

参数说明
ServiceNamespace服务命名空间,取值见下方枚举
ResourceId资源标识,格式因服务而异(详见 3.3)
ScalableDimension可扩展维度,标识要调整的容量属性
MinCapacity/MaxCapacity容量下限/上限(整数)
RoleARN扩缩容时使用的 IAM 角色 ARN
SuspendedState暂停状态,可包含DynamicScalingInSuspended、DynamicScalingOutSuspended、ScheduledScalingSuspended

支持的服务命名空间(定义于 models.py):

appstream, rds, lambda, cassandra, dynamodb, custom-resource, elasticmapreduce, ec2, comprehend, ecs, sagemaker, kafka

支持的典型可扩展维度(部分,完整枚举见 models.py):

维度值适用服务
ecs:service:DesiredCountECS 服务期望实例数
dynamodb:table:ReadCapacityUnits/WriteCapacityUnitsDynamoDB 表
dynamodb:index:ReadCapacityUnits/WriteCapacityUnitsDynamoDB GSI
rds:cluster:ReadReplicaCount/rds:cluster:CapacityRDS 集群
lambda:function:ProvisionedConcurrencyLambda 预置并发
sagemaker:variant:DesiredInstanceCountSageMaker 变体实例数
ec2:spot-fleet-request:TargetCapacityEC2 Spot Fleet
appstream:fleet:DesiredCapacityAppStream 队列
elasticmapreduce:instancefleet:*/instancegroup:InstanceCountEMR
comprehend:document-classifier-endpoint:DesiredInferenceUnitsComprehend
cassandra:table:ReadCapacityUnits/WriteCapacityUnitsKeyspaces
kafka:broker-storage:VolumeSizeMSK broker 存储
custom-resource:ResourceType:Property自定义资源
基本示例(ECS)
import boto3 from moto import mock_aws @mock_aws def test_register_ecs_target(): ecs = boto3.client("ecs", region_name="us-east-1") ecs.create_cluster(clusterName="default") ecs.register_task_definition( family="my-task", containerDefinitions=[ {"name": "hello", "image": "docker/hello-world:latest", "cpu": 1024, "memory": 400, "essential": True} ], ) ecs.create_service( cluster="default", serviceName="sample-webapp", taskDefinition="my-task", desiredCount=2, ) client = boto3.client("application-autoscaling", region_name="us-east-1") resp = client.register_scalable_target( ServiceNamespace="ecs", ResourceId="service/default/sample-webapp", ScalableDimension="ecs:service:DesiredCount", MinCapacity=1, MaxCapacity=10, RoleARN="arn:aws:iam::123456789012:role/ecs-autoscale", SuspendedState={ "DynamicScalingInSuspended": True, "DynamicScalingOutSuspended": False, "ScheduledScalingSuspended": False, }, ) assert resp["ScalableTargetARN"].startswith( "arn:aws:application-autoscaling:us-east-1:" )

注意:在 Moto 中注册 ECS 目标前,必须先创建对应的 ECS 服务。后端会调用_ecs_service_exists_for_target检查服务是否存在,否则抛出ValidationException: ECS service doesn't exist: ...(见 models.py)。对应的测试用例位于 tests/test_applicationautoscaling/test_validation.py。

重复注册同一目标时,行为是更新:register_scalable_target会先调用_scalable_target_exists判断是否已存在,若存在则仅更新min_capacity/max_capacity/suspended_state(见 models.py)。验证该行为的测试见 tests/test_applicationautoscaling/test_applicationautoscaling.py。

3.2 describe_scalable_targets:查询可扩展目标

支持按ServiceNamespace、ResourceIds列表与ScalableDimension过滤。从源码看(models.py):

  • 先按命名空间展平全部目标(_flatten_scalable_targets);
  • 若传了ScalableDimension,只保留维度匹配者;
  • 若传了ResourceIds,只保留 ID 命中者。

响应层(responses.py)默认MaxResults=50,并支持NextToken分页翻页,返回字段包含CreationTime、MaxCapacity、MinCapacity、ResourceId、RoleARN、ScalableDimension、ServiceNamespace、ScalableTargetARN、SuspendedState。

示例
resp = client.describe_scalable_targets(ServiceNamespace="ecs") for t in resp["ScalableTargets"]: print(t["ResourceId"], t["ScalableDimension"], t["MinCapacity"], t["MaxCapacity"])

3.3 ResourceId 的格式差异

Moto 的_get_resource_type_from_resource_id(models.py)与_target_params_are_valid(models.py)共同保证了 resource_id、namespace、dimension 三者的一致性校验。其规则是:dimension形如namespace:resource_type:属性,其中resource_type通常取自resource_id的第一段,但存在例外:

  • SageMaker 端点:endpoint/MyEndPoint/variant/MyVariant,resource_type 为第三段variant;
  • DynamoDB GSI:table/my-table/index/my-table-index,resource_type 为第三段index;
  • Keyspaces 表:keyspace/mykeyspace/table/mytable,resource_type 为第三段table;
  • Comprehend:直接使用 ARN,resource_type 取 ARN 的最后一个:分段;
  • MSK:kafka:broker-storage属于例外,resource_type 不取自 resource_id。

以下组合在测试中被验证有效(见 tests/test_applicationautoscaling/test_applicationautoscaling.py):

# (namespace, resource_id, scalable_dimension) ("ecs", "service/default/sample-webapp", "ecs:service:DesiredCount") ("ec2", "spot-fleet-request/sfr-73fbd2ce-aa30-494c-8788-1cee4EXAMPLE", "ec2:spot-fleet-request:TargetCapacity") ("elasticmapreduce", "instancegroup/j-2EEZNYKUA1NTV/ig-1791Y4E1L8YI0", "elasticmapreduce:instancegroup:InstanceCount") ("appstream", "fleet/sample-fleet", "appstream:fleet:DesiredCapacity") ("dynamodb", "table/my-table", "dynamodb:table:ReadCapacityUnits") ("dynamodb", "table/my-table/index/my-table-index", "dynamodb:index:ReadCapacityUnits") ("rds", "cluster:my-db-cluster", "rds:cluster:ReadReplicaCount") ("sagemaker", "endpoint/MyEndPoint/variant/MyVariant", "sagemaker:variant:DesiredInstanceCount") ("comprehend", "arn:aws:comprehend:us-west-2:123456789012:document-classifier-endpoint/EXAMPLE", "comprehend:document-classifier-endpoint:DesiredInferenceUnits") ("lambda", "function:my-function:prod", "lambda:function:ProvisionedConcurrency") ("cassandra", "keyspace/mykeyspace/table/mytable", "cassandra:table:ReadCapacityUnits") ("custom-resource", "https://test-endpoint.amazon.com/ScalableDimension/test-resource", "custom-resource:ResourceType:Property")

3.4 deregister_scalable_target:注销目标

删除指定目标;若目标不存在,则抛出ValidationException: No scalable target found for service namespace: ..., resource ID: ..., scalable dimension: ...(见 models.py)。

client.deregister_scalable_target( ServiceNamespace="ecs", ResourceId="service/default/sample-webapp", ScalableDimension="ecs:service:DesiredCount", )

四、扩缩容策略(Scaling Policy)的完整生命周期

4.1 put_scaling_policy:创建或更新策略

支持StepScaling与TargetTrackingScaling两种策略类型,分别通过StepScalingPolicyConfiguration与TargetTrackingScalingPolicyConfiguration传入策略体。

策略类型配置结构示例
TargetTrackingScaling{"TargetValue": 70.0, "PredefinedMetricSpecification": {"PredefinedMetricType": "DynamoDBReadCapacityUtilization"}}
StepScaling{"AdjustmentType": "ChangeInCapacity", "StepAdjustments": [{"ScalingAdjustment": 10}], "MinAdjustmentMagnitude": 2}

在 models.py 中,FakeApplicationAutoscalingPolicy会根据policy_type将策略体分别存入step_scaling_policy_configuration或target_tracking_scaling_policy_configuration;传入其他类型(如PredictiveScaling或非法字符串)会抛出:

1 validation error detected: Value '{policy_type}' at 'policyType' failed to satisfy constraint: Member must satisfy enum value set: [PredictiveScaling, StepScaling, TargetTrackingScaling]

该行为由 tests/test_applicationautoscaling/test_applicationautoscaling_policies.py 验证。

关键联动:自动创建 CloudWatch 告警。对于TargetTrackingScaling策略,Moto 会自动在 CloudWatch 后端创建告警(create_alarms,见 models.py):

  • 命名空间dynamodb:创建 4 个告警(AlarmHigh/AlarmLow/ProvisionedCapacityHigh/ProvisionedCapacityLow,监控ConsumedReadCapacityUnits与ProvisionedReadCapacityUnits,见 models.py);
  • 命名空间ecs:创建 2 个告警,依据PredefinedMetricType是否包含Memory选择MemoryUtilization或CPUUtilization(见 models.py)。

这些告警会出现在put_scaling_policy的响应Alarms字段中,也可通过describe_scaling_policies查询到,并能通过 CloudWatch 的describe_alarms直接读取——测试用例 tests/test_applicationautoscaling/test_applicationautoscaling_policies.py 对此做了交叉验证。

client.put_scaling_policy( PolicyName="dynamodb-scale-out", ServiceNamespace="dynamodb", ResourceId="table/my-table", ScalableDimension="dynamodb:table:ReadCapacityUnits", PolicyType="TargetTrackingScaling", TargetTrackingScalingPolicyConfiguration={ "TargetValue": 70.0, "PredefinedMetricSpecification": { "PredefinedMetricType": "DynamoDBReadCapacityUtilization" }, }, )

4.2 describe_scaling_policies:查询策略

支持按ServiceNamespace(必填)、PolicyNames、ResourceId、ScalableDimension过滤,并支持MaxResults与NextToken分页(默认每页 100,见 models.py)。

返回字段包括PolicyARN、PolicyName、ServiceNamespace、ResourceId、ScalableDimension、PolicyType、CreationTime、Alarms,并依据策略类型附带StepScalingPolicyConfiguration或TargetTrackingScalingPolicyConfiguration(见 responses.py)。

策略 ARN 格式为:

arn:aws:autoscaling:{region}:{account_id}:scalingPolicy:{uuid}:resource/{namespace}/{resource_id}:policyName/{policy_name}

4.3 delete_scaling_policy:删除策略

按PolicyName、ServiceNamespace、ResourceId、ScalableDimension定位并删除;删除时会同步清理自动创建的 CloudWatch 告警(delete_alarms,见 models.py)。若策略不存在,抛出ValidationException: No scaling policy found for ...。

五、定时操作(Scheduled Action)的使用

5.1 put_scheduled_action:创建定时操作

用于按Schedule(cron 或 rate 表达式)定期调整目标容量。参数:

参数说明
ServiceNamespace服务命名空间
ScheduledActionName操作名称
ResourceId目标资源
ScalableDimension维度
Schedule如rate(2 minutes)、cron(0 18 * * ? *)
Timezone时区(如UTC)
StartTime/EndTime生效起止时间
ScalableTargetAction包含MinCapacity/MaxCapacity的容量动作

示例(基于 tests/test_applicationautoscaling/test_applicationautoscaling.py):

client.register_scalable_target( ServiceNamespace="dynamodb", ResourceId="table/my-table", ScalableDimension="dynamodb:table:ReadCapacityUnits", MinCapacity=1, MaxCapacity=100, ) client.put_scheduled_action( ServiceNamespace="dynamodb", Schedule="rate(2 minutes)", ScheduledActionName="action_name", ResourceId="table/my-table", ScalableDimension="dynamodb:table:ReadCapacityUnits", ScalableTargetAction={"MinCapacity": 1, "MaxCapacity": 5}, )

重复对相同(namespace, action_name, resource_id, dimension)调用put_scheduled_action会更新既有操作而非新增(见 models.py 及测试 tests/test_applicationautoscaling/test_applicationautoscaling.py)。

生成的操作 ARN 格式为:

arn:aws:autoscaling:{region}:{account_id}:scheduledAction:{namespace}/{resource_id}:scheduledActionName/{action_name}

注意:Moto 仅记录定时操作元数据,并不会真的在调度时刻触发扩缩容动作。

5.2 describe_scheduled_actions:查询定时操作

按ServiceNamespace(必填)过滤,可选ScheduledActionNames、ResourceId、ScalableDimension进一步筛选。源码中明确注释Pagination is not yet implemented(见 models.py)。

返回字段:ScheduledActionName、ScheduledActionARN、ServiceNamespace、Schedule、Timezone、ResourceId、ScalableDimension、StartTime、EndTime、CreationTime、ScalableTargetAction。

5.3 delete_scheduled_action:删除定时操作

按四个维度定位并删除;若不存在则静默成功(不抛错,见 models.py)。

六、参数校验规则与常见报错

6.1 双层级联校验

Moto 对该服务实现了两道校验:

  1. 响应层校验(_validate_params,见 responses.py):在分发前校验ServiceNamespace与ScalableDimension是否属于合法枚举,非法时抛出形如1 validation error detected: Value 'foo' at 'serviceNamespace' failed to satisfy constraint: ...的ValidationException(HTTP 400);
  2. 模型层校验(_target_params_are_valid,见 models.py):注册/更新目标时校验命名空间、维度与 resource_id 的一致性(如ecs命名空间必须配ecs:service:DesiredCount维度和service/...形式的资源 ID)。

6.2 常见错误对照

场景错误码说明
非法ScalableDimensionValidationException如"foo",报 1 个校验错误
非法ServiceNamespaceValidationException如"foo",报 1 个校验错误
两者均非法ValidationException报 2 个校验错误(2 validation errors detected)
ECS 服务不存在ClusterNotFoundException注册 ECS 目标时底层 ECS 后端抛出的Cluster not found.
删除不存在的目标/策略ValidationException如No scalable target found for .../No scaling policy found for ...
非法 PolicyTypeValidationException枚举限制为[PredictiveScaling, StepScaling, TargetTrackingScaling]

相关验证测试集中在 tests/test_applicationautoscaling/test_validation.py。

七、与其他服务的联动与注意事项

7.1 与 ECS 的联动

后端初始化时即持有同账号同区域的 ECS 后端引用(self.ecs_backend = ecs_backends[account_id][region_name],见 models.py)。这意味着:

  • 注册 ECS 目标前必须先存在对应的 ECS 服务(否则报ECS service doesn't exist);
  • TargetTracking 策略会自动为 ECS 服务创建基于CPUUtilization/MemoryUtilization的 CloudWatch 告警,告警维度为ClusterName与ServiceName。

7.2 与 CloudWatch 的联动

FakeApplicationAutoscalingPolicy通过cloudwatch_backends[account_id][region_name]直接操作 CloudWatch 后端,因此在@mock_aws上下文中创建策略后,可以立刻通过 CloudWatch 客户端查询到对应告警;删除策略时告警也随之删除。这在测试真实告警依赖逻辑时非常有用。

7.3 分页实现的现状

文档声明分页未实现,但源码表明两个查询接口已具备分页能力:

  • describe_scalable_targets:响应层默认MaxResults=50,支持NextToken(responses.py),测试见 tests/test_applicationautoscaling/test_applicationautoscaling.py;
  • describe_scaling_policies:后端默认每页 100 条并返回NextToken(models.py)。

而describe_scheduled_actions确实未实现分页。

7.4 区域与账号隔离

后端通过BackendDict(ApplicationAutoscalingBackend, "application-autoscaling")注册(models.py),数据按account_id + region隔离。使用不同 region 的客户端(如eu-west-1、ap-southeast-1)看到的将是彼此独立的数据集合,测试中对此有覆盖(如 tests/test_applicationautoscaling/test_applicationautoscaling.py)。

八、测试环境快速上手

8.1 启用方式

在测试中使用@mock_aws装饰器即可,无需任何额外配置:

from moto import mock_aws import boto3 @mock_aws def test_auto_scaling_flow(): # ... 创建 ECS 服务 / DynamoDB 表 ... client = boto3.client("application-autoscaling", region_name="us-east-1") client.register_scalable_target(...) client.put_scaling_policy(...) client.put_scheduled_action(...) assert len(client.describe_scalable_targets(ServiceNamespace="ecs")["ScalableTargets"]) == 1

8.2 既有测试参考

仓库中已有完整测试可作参考:

  • tests/test_applicationautoscaling/test_applicationautoscaling.py:目标注册/查询/更新、策略、定时操作及分页;
  • tests/test_applicationautoscaling/test_applicationautoscaling_policies.py:策略类型校验与 CloudWatch 告警联动;
  • tests/test_applicationautoscaling/test_validation.py:参数校验行为。

其中application_autoscaling_aws_verified装饰器(tests/test_applicationautoscaling/init.py)允许在设置MOTO_TEST_ALLOW_AWS_REQUEST=true时直接对真实 AWS 运行验证性测试,未设置时则在mock_aws上下文中运行,可作为与真实行为对照的依据。

8.3 局限提醒

在 Moto 中使用该服务时请留意以下限制:

  • describe_scaling_activities、get_predictive_scaling_forecast、标签相关操作未实现;
  • put_scaling_policy不支持PredictiveScaling类型;
  • 定时操作只记录元数据,不会在计划时间点真实触发容量变更;
  • TargetTracking 策略创建的 CloudWatch 告警为 Moto 自动生成的模拟告警,其阈值(如 DynamoDB 的 42.0/30.0)为内部实现值,不应将其当作真实 AWS 的告警阈值。

九、小结

Moto 对 AWS Application Auto Scaling 的模拟覆盖了核心闭环:注册可扩展目标 → 配置扩缩容策略 → 创建定时操作 → 查询与清理,并在 ECS、DynamoDB 场景下与 CloudWatch 告警产生真实联动,足以支撑大多数本地与 CI 场景下的扩缩容逻辑测试。尚未实现的接口(活动历史、预测式扩缩容、标签管理)在使用时应提前规避。结合本文给出的参数表格、资源 ID 格式与测试用例,你可以快速在自己的项目中复现并验证 Auto Scaling 相关行为。

  • Mock
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载
上一篇:SVGnest终极指南:材料切割优化利器完全解析
下一篇:terraform-provider-aws 数据源实战:用 aws_connect_vocabulary 查询 Amazon Connect 自定义词汇表

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询