☰
Moto 中 Amazon Managed Prometheus(amp)服务的模拟实现与实战指南
2026/9/25 8:18:26 网站建设 项目流程
  • Mock
  • 测试

【免费下载链接】moto

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

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

Amazon Managed Prometheus(AMP,AWS 的托管 Prometheus 服务)是云上运行 Prometheus 工作负载的核心托管组件。本文聚焦开源仓库 moto 中amp服务的模拟实现,以 docs/docs/services/amp.rst 中列出的功能清单为骨架,结合 moto/amp/models.py 等源码与 tests/test_amp/ 下的测试用例,全面讲解 Workspace、Rule Groups Namespace、Logging Configuration、标签管理与分页等已实现 API 的用法、参数细节与底层原理。读完本文,你将能在本地测试环境中用mock_aws装饰器完整模拟 AMP 工作流的创建、查询、更新与删除操作。

AMP 服务在 Moto 中的整体支持范围

AMP 服务在 moto 中对应moto/amp模块,其网络入口模拟的是 AWS AMP 的真实端点aps.{region}.amazonaws.com,这一点可以从 moto/amp/urls.py 的url_bases定义得到确认:

url_bases = [ r"https?://aps\.(.+)\.amazonaws\.com", ]

核心后端类为 moto/amp/models.py 中的PrometheusServiceBackend,它继承自BaseBackend,并通过BackendDict(PrometheusServiceBackend, "amp")按账户(account)+ 区域(region)隔离存储,因此不同 region 的 workspace 互不干扰,这与 moto 其他服务的多区域模型一致。

已实现功能清单

原文档通过勾选列表标注了每个 API 的实现状态,这是判断 moto 中 amp 服务可用能力的最权威依据。已实现([X])的接口如下:

API 操作说明ClientToken 支持
create_workspace创建工作区未实现
describe_workspace查询工作区详情—
list_workspaces列出工作区(支持分页与 alias 过滤)—
update_workspace_alias更新工作区别名未实现
delete_workspace删除工作区未实现
create_rule_groups_namespace创建规则组命名空间未实现
describe_rule_groups_namespace查询规则组命名空间—
list_rule_groups_namespaces列出规则组命名空间(支持分页与名称前缀过滤)—
put_rule_groups_namespace更新规则组命名空间内容未实现
delete_rule_groups_namespace删除规则组命名空间未实现
create_logging_configuration配置工作区日志输出—
describe_logging_configuration查询日志配置—
update_logging_configuration更新日志配置—
delete_logging_configuration删除日志配置—
tag_resource为资源打标签—
untag_resource移除资源标签—
list_tags_for_resource查询资源标签—

同时,原文档也明确标注了未实现的操作,包括:create_alert_manager_definition、delete_alert_manager_definition、describe_alert_manager_definition、put_alert_manager_definition(Alert Manager 定义相关全套)、create_scraper、delete_scraper、describe_scraper、list_scrapers、update_scraper等 Scraper 相关接口,以及create_anomaly_detector、delete_anomaly_detector、describe_anomaly_detector、list_anomaly_detectors、put_anomaly_detector(异常检测器)、查询日志配置(create/describe/update/delete_query_logging_configuration)、资源策略(describe/delete/put_resource_policy)、工作区配置(describe/update_workspace_configuration)与get_default_scraper_configuration。在使用这些未实现的 API 时,boto3 客户端会抛出服务端错误,这一点需要在使用前评估。

关于 ClientToken 参数的特别说明

原文档在多个已实现 API 下方反复标注:

The ClientToken-parameter is not yet implemented

即create_workspace、create_rule_groups_namespace、delete_rule_groups_namespace、put_rule_groups_namespace、delete_workspace、update_workspace_alias等接口虽然在签名上接受clientToken参数(调用时不会报错),但该参数在模拟实现中不会产生任何幂等性语义。这一点在 moto/amp/models.py 中可以直接看到——create_workspace的 docstring 即为该句说明,方法体内并未读取 clientToken。测试文件 tests/test_amp/test_amp_workspaces.py 中create_workspace(alias="test", clientToken="mytoken")的写法也证实了该参数可被传入但被忽略。

Workspace 生命周期管理:创建、查询、更新与删除

Workspace 是 AMP 的核心资源,对应源码中的 Workspace 模型。模拟实现会在创建时生成以下关键字段:

  • workspace_id:形如ws-{uuid4}的随机 ID(来自moto.moto_api._internal.mock_random.uuid4);
  • arn:形如arn:{partition}:aps:{region}:{account_id}:workspace/{workspace_id},其中 partition 由get_partition(region)计算(如aws、aws-cn等);
  • prometheusEndpoint:形如https://aps-workspaces.{region}.amazonaws.com/workspaces/{workspace_id}/;
  • status:固定为{"statusCode": "ACTIVE"};
  • createdAt:创建时的 unix 时间戳。

创建与查询

使用mock_aws装饰器即可在本地模拟完整的 AWS 调用链:

import boto3 from moto import mock_aws @mock_aws def test_workspace_lifecycle(): client = boto3.client("amp", region_name="ap-southeast-1") resp = client.create_workspace(alias="my-prom", tags={"env": "dev"}) workspace_id = resp["workspaceId"] assert resp["status"] == {"statusCode": "ACTIVE"} assert resp["alias"] == "my-prom" detail = client.describe_workspace(workspaceId=workspace_id)["workspace"] assert detail["workspaceId"] == workspace_id assert "prometheusEndpoint" in detail assert "createdAt" in detail

注意响应结构的差异:create_workspace直接返回 workspace 对象的所有字段(顶层展开),而describe_workspace返回嵌套的{"workspace": {...}}结构,与真实 AWS API 保持一致(见 moto/amp/responses.py)。

列出与别名更新

list_workspaces支持两个筛选维度:

  • alias:精确匹配过滤,仅返回alias完全相等的 workspace(见 models.py 中的w.alias == alias);
  • maxResults/nextToken:分页参数,默认每页 100 条。
client.update_workspace_alias(alias="renamed", workspaceId=workspace_id) # alias 过滤 only_renamed = client.list_workspaces(alias="renamed")["workspaces"] assert len(only_renamed) == 1

删除与异常行为

delete_workspace直接从内部 dict 中弹出该 workspace(pop(workspace_id, None)),随后对该 ID 执行describe_workspace会抛出ResourceNotFoundException。测试 test_amp_workspaces.py 验证了这一点:

client.delete_workspace(workspaceId=workspace_id) with pytest.raises(ClientError) as exc: client.describe_workspace(workspaceId=workspace_id) assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException" assert exc.value.response["Error"]["Message"] == "Workspace not found"

该异常由 moto/amp/exceptions.py 中的WorkspaceNotFound抛出,HTTP 状态码为 404,resourceType为AWS::APS::Workspace。

Rule Groups Namespace:Prometheus 规则组的命名空间管理

Rule Groups Namespace 用于承载 Prometheus 的 recording rule 与 alerting rule 配置,是 AMP 模拟中最常用的"内容型"资源。每个命名空间归属于某个 workspace,对应 RuleGroupNamespace 模型,其 ARN 格式为:

arn:{partition}:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{name}

命名空间模型内部维护created_at与modified_at两个时间戳,put_rule_groups_namespace更新data内容时仅刷新modified_at(见 update 方法)。

创建与描述

data参数在 boto3 中为 bytes 类型(底层为 YAML 格式的规则组内容):

@mock_aws def test_rulegroups(): client = boto3.client("amp", region_name="eu-west-1") workspace_id = client.create_workspace()["workspaceId"] client.create_rule_groups_namespace( data=b"groups:\n - name: example\n rules: []", name="my-rules", workspaceId=workspace_id, ) ns = client.describe_rule_groups_namespace( name="my-rules", workspaceId=workspace_id )["ruleGroupsNamespace"] assert ns["data"] == b"groups:\n - name: example\n rules: []" assert ns["status"] == {"statusCode": "ACTIVE"} assert "createdAt" in ns and "modifiedAt" in ns

与 workspace 一样,创建/更新操作直接返回命名空间对象字段,而describe_rule_groups_namespace返回嵌套的{"ruleGroupsNamespace": {...}}结构。

更新(put)与删除

client.put_rule_groups_namespace( name="my-rules", workspaceId=workspace_id, data=b"groups: []" ) ns = client.describe_rule_groups_namespace( name="my-rules", workspaceId=workspace_id )["ruleGroupsNamespace"] assert ns["data"] == b"groups: []" client.delete_rule_groups_namespace(name="my-rules", workspaceId=workspace_id)

删除后查询同一命名空间会抛出RuleGroupNamespaceNotFound(404,resourceType为AWS::APS::RuleGroupNamespace,见 exceptions.py)。

列表与名称前缀过滤

list_rule_groups_namespaces的name参数是前缀匹配而非精确匹配——源码中通过ns_name.startswith(name)实现(models.py)。测试 test_amp_rulegroupnamespaces.py 演示了典型行为:存在ns0~ns14共 15 个命名空间时,name="ns1"会匹配到ns1、ns10~ns14共 6 个。

Logging Configuration:工作区日志输出配置

日志配置用于将 AMP 的查询与指标日志发送到 CloudWatch Logs,对应 models.py 中的四个方法。模拟实现的数据结构包含logGroupArn、createdAt、status(固定ACTIVE)与workspace字段。

创建与查询

@mock_aws def test_logging(): client = boto3.client("amp", region_name="us-east-2") workspace_id = client.create_workspace()["workspaceId"] resp = client.create_logging_configuration( workspaceId=workspace_id, logGroupArn="arn:aws:logs:us-east-2:123456789012:log-group:/amp/logs" ) assert resp["status"] == {"statusCode": "ACTIVE"} cfg = client.describe_logging_configuration(workspaceId=workspace_id)["loggingConfiguration"] assert cfg["logGroupArn"].endswith("/amp/logs") assert cfg["workspace"] == workspace_id

值得注意的细节:在从未创建过日志配置时,describe_logging_configuration返回的是空字典{}(models.py 中logging_config is None时返回{}),而不是报错。测试 test_amp_logging_config.py 专门验证了这一行为。

更新与删除

client.update_logging_configuration( workspaceId=workspace_id, logGroupArn="arn:aws:logs:us-east-2:123456789012:log-group:/amp/logs-v2" ) cfg = client.describe_logging_configuration(workspaceId=workspace_id)["loggingConfiguration"] assert "modifiedAt" in cfg assert cfg["logGroupArn"].endswith("/amp/logs-v2") client.delete_logging_configuration(workspaceId=workspace_id) assert client.describe_logging_configuration(workspaceId=workspace_id)["loggingConfiguration"] == {}

update_logging_configuration会覆盖logGroupArn并追加modifiedAt时间戳;delete_logging_configuration将内部配置置为None,使后续查询重新回到空字典状态。

标签管理:tag_resource / untag_resource / list_tags_for_resource

AMP 模拟的标签能力统一由TaggingService提供(models.py),可作用于 Workspace 与 RuleGroupNamespace 两类资源。标签以 ARN 为键存储,并通过tag_fn回调挂接到每个模型上,因此describe返回的响应中会直接携带tags字段。

@mock_aws def test_tags(): client = boto3.client("amp", region_name="us-east-2") arn = client.create_workspace(alias="test", tags={"env": "prod"})["arn"] client.tag_resource(resourceArn=arn, tags={"team": "platform"}) assert client.list_tags_for_resource(resourceArn=arn)["tags"] == { "env": "prod", "team": "platform" } client.untag_resource(resourceArn=arn, tagKeys=["env"]) assert client.list_tags_for_resource(resourceArn=arn)["tags"] == {"team": "platform"}

由于底层是统一的TaggingService,Workspace 与 RuleGroupNamespace 的标签行为完全一致(两组测试文件分别验证了各自的场景)。untag_resource通过untag_resource_using_names按标签名批量删除,可一次传入多个 tagKey。

分页行为:默认页大小与 nextToken

AMP 的两个列表接口(list_workspaces、list_rule_groups_namespaces)都支持 AWS 风格的分页,分页模型定义在 moto/amp/utils.py:

PAGINATION_MODEL = { "list_workspaces": { "input_token": "next_token", "limit_key": "max_results", "limit_default": 100, "unique_attribute": "arn", }, "list_rule_groups_namespaces": { "input_token": "next_token", "limit_key": "max_results", "limit_default": 100, "unique_attribute": "name", }, }

关键行为:

  • 默认页大小 100:不传maxResults时,超过 100 条的列表只返回前 100 条并附带nextToken;
  • maxResults可缩小或扩大:测试中展示了maxResults=15搭配nextToken逐页取数,以及maxResults=1000一次性取完的场景(test_amp_workspaces.py);
  • 翻页游标属性:workspace 分页以arn为唯一游标属性,rulegroupsnamespace 分页以name为唯一游标属性;
  • 最后一页无nextToken:取完所有数据后响应中不再包含nextToken字段(test_amp_rulegroupnamespaces.py)。

分页逻辑由moto.utilities.paginator中的paginate装饰器实现,与 moto 其他服务保持一致。

请求路由与响应分发机制

理解 amp 的模拟实现,还需要看清"HTTP 请求 → 后端方法"的调用链。整个链路由三部分构成:

  1. URL 路由(moto/amp/urls.py):将所有aps.{region}.amazonaws.com下的路径统一派发到PrometheusServiceResponse.dispatch,包括/workspaces、/workspaces/{id}/alias、/workspaces/{id}/logging、/workspaces/{id}/rulegroupsnamespaces以及/tags/...系列路径;
  2. 响应分发(moto/amp/responses.py):PrometheusServiceResponse继承BaseResponse,根据请求中携带的 API action 名调用对应的处理函数(如create_workspace、describe_workspace等),处理函数负责从请求 body/路径中解析参数、调用 backend 方法、并把结果序列化为与真实 AWS 一致的 JSON 结构;
  3. 后端模型(moto/amp/models.py):PrometheusServiceBackend持有self.workspaces字典与self.tagger,通过amp_backends[current_account][region]获取当前账户与区域对应的后端实例,实现数据隔离。

路径参数的解析方式值得注意:responses.py中大量使用self.path.split("/")配合unquote从 URL 中提取workspace_id、name、resource_arn等参数,例如describe_workspace取路径最后一段作为 workspaceId、create_rule_groups_namespace取倒数第二段。

使用限制与注意事项总结

综合原文档标注与源码实现,使用 moto 的 amp 模拟时有以下几点需要提前知悉:

  • ClientToken 参数仅占位:create_workspace、create_rule_groups_namespace、put_rule_groups_namespace、delete_rule_groups_namespace、delete_workspace、update_workspace_alias均接受但不处理clientToken,不会提供幂等去重语义;
  • Alert Manager / Scraper / Anomaly Detector 等 API 未实现:涉及告警管理、抓取任务、异常检测的业务场景目前无法在 moto 中端到端模拟,需要等待后续版本补齐;
  • 状态字段简化:workspace 与命名空间的status固定为{"statusCode": "ACTIVE"},不会模拟CREATING、DELETING、UPDATING等中间状态;
  • 数据仅存在于内存:所有 workspace、命名空间与标签均保存在后端实例的内存字典中,进程退出即丢失,适合测试场景而非持久化需求;
  • 区域隔离:由于BackendDict按 account+region 建实例,同一测试中跨 region 创建的 workspace 彼此独立。

运行 amp 相关测试

仓库在 tests/test_amp/ 下提供了三个测试文件,分别覆盖 workspace(test_amp_workspaces.py)、rule group namespaces(test_amp_rulegroupnamespaces.py)与日志配置(test_amp_logging_config.py)。在仓库根目录执行以下命令即可验证 amp 模拟的全部已实现功能:

pytest tests/test_amp/

这些测试既是实现行为的权威佐证,也是撰写自定义测试时可直接参考的模板——它们展示了mock_aws装饰器、boto3 客户端的 region 选择(如ap-southeast-1、eu-west-1、us-east-2)、以及ClientError断言的标准写法。总体而言,moto 对 AMP 的模拟已经覆盖了工作区生命周期、规则组命名空间、日志配置与标签管理这四类最核心的日常操作,足以支撑大多数基于 AMP 的应用在本地测试环境中的行为验证。

  • Mock
  • 测试

【免费下载链接】moto

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

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载
上一篇:CommandLineParser选项组详解:互斥选项与分组管理的完整指南
下一篇:神经网络从零实现:Machine Learning with PyTorch and Scikit-Learn深度学习基础

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

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

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

立即咨询