首页
/ Sentry Hybrid Cloud RPC 服务开发指南：跨 Silo 服务的创建、演进与安全下线

Sentry Hybrid Cloud RPC 服务开发指南：跨 Silo 服务的创建、演进与安全下线

2026-09-08 21:49:26作者：房伟宁
sentry
Developer-first error tracking and performance monitoring
项目地址：https://gitcode.com/GitHub_Trending/sen/sentry

Sentry 的混合云（Hybrid Cloud）架构将系统拆分为 Control Silo（用户、认证、组织管理）与 Cell Silo（项目、事件、Issue、计费等数据面），RPC 服务层是二者之间唯一的跨数据库通信桥梁。本文以仓库中的开发技能文档 hybrid-cloud-rpc/SKILL.md 为主线，结合其四份配套参考（service-template.md、rpc-models.md、resolvers.md、deprecation.md）与 RPC 框架源码，完整讲解如何在 Sentry 中新建 RPC 服务、设计 RpcModel、选择 Cell 解析策略、给方法做向后兼容的签名演进，以及按三阶段流程安全地弃用和删除远端方法。读完你将掌握一套可直接套用的工程规范与测试方法，能够独立完成"新增 RPC 方法 / 创建 RPC 服务 / 弃用 RPC 方法 / 跨 Silo 服务开发"的完整闭环。

Hybrid Cloud 背景：为什么需要 RPC 服务

在 Sentry 的多 Silo 部署模型中：

  • Control Silo 承载用户认证、组织归属映射等全局性数据；
  • Cell Silo 承载项目、事件、Issue、计费等数据面，一个组织通过 OrganizationMapping 归属到某个具体 Cell。

RpcService 正是为打通这两类数据而生的抽象层。从源码看，一个 RPC 服务由两部分构成：

  1. Base Service（抽象接口）：继承 RpcService 的抽象类，声明 key（服务在 URL 中的 slug）与 local_mode（该服务使用本地实现的 Silo 模式）；
  2. Remote Delegate / DatabaseBacked 实现：create_delegation() 会根据当前进程的 SiloMode 自动选择——MONOLITH 模式与 local_mode 匹配时使用本地数据库实现，其他模式则构造一个 Remote 代理类，把调用序列化为 HTTP 请求发给目标 Silo（见 create_delegation 与 _create_remote_implementation）。

每次方法调用都会经过 RpcMethodSignature 完成参数序列化；若服务运行于 CELL 模式，还必须先解析出目标 Cell 再派发。

硬性约束：先把规则背下来

整个技能文档以一组"Critical Constraints"开头，这些约束直接由框架的反射式设计决定，违反任意一条都会导致线上序列化静默失败或启动即报错：

NEVER 在 service.py 或 model.py 中使用 from __future__ import annotations——RPC 框架在 import 时会反射类型注解，前向引用（字符串注解）会静默破坏序列化。

ALL RPC 方法参数必须是 keyword-only（签名中使用 *）。

ALL 参数与返回类型必须写全类型注解，禁止字符串前向引用。

ONLY 允许可序列化类型：int、str、bool、float、None、Optional[T]、list[T]、dict[str, T]、RpcModel 子类、Enum 子类、datetime.datetime。

服务 MUST 位于 12 个已注册的发现包（discovery packages）之一。

对敏感字段（token、密钥、配置 blob、metadata 字典）使用 Field(repr=False)，防止其泄漏进日志与错误报告，完整规则见 rpc-models.md。

这些约束与源码实现一一对应：RpcService.__init_subclass__ 会校验 key 与 local_mode 必须存在（否则抛 RpcServiceSetupException），_get_and_validate_local_implementation 会用 inspect.signature 逐一核对实现方法的参数名与抽象方法完全一致（service.py），注解反射则依赖 SerializableFunctionSignature（sig.py）在 import 时即时解析真实类型对象。

第一步：判断本次操作类型

不同诉求对应不同流程，先分类再动手：

意图 流程
创建一个全新的 RPC 服务 先看第 2 步，再执行第 3 步
给已有服务添加方法 先看第 2 步，再执行第 4 步
修改已有方法的签名 执行签名演进流程
弃用或删除方法 / 服务 执行三阶段弃用流程

第二步：确定 Silo 模式

服务的 local_mode 决定了数据库实现的运行位置，也决定了方法上需要挂哪种装饰器：

数据所在位置 local_mode 方法装饰器 仓库实例
Cell Silo（项目、事件、Issue、组织数据、计费） SiloMode.CELL @cell_rpc_method(resolve=...) OrganizationService（见 organization/service.py）
Control Silo（用户、认证、组织映射） SiloMode.CONTROL @rpc_method OrganizationMemberMappingService

决策规则：你需要的 Django 模型在 Cell 数据库就选 SiloMode.CELL，在 Control 数据库就选 SiloMode.CONTROL。注意框架源码会在启动期强制校验这一点——若在 CELL 服务上漏写 @cell_rpc_method，或在非 CELL 服务上误用 @cell_rpc_method，都会直接抛出 RpcServiceSetupException（见 service.py）。

Cell Silo 服务有额外要求：每个 RPC 方法都必须携带一个 CellResolutionStrategy（通过 @cell_rpc_method(resolve=...) 声明），否则框架不知道远端调用该路由到哪个 Cell。完整的 resolver 对照表见 resolvers.md。

第三步：创建新服务

目录结构规范

每个 RPC 服务对应一个五文件包，service-template.md 提供了全部可直接复制的模板：

src/sentry/{domain}/services/{service_name}/
├── __init__.py      # 重导出 model 和 service
├── model.py         # RpcModel 子类（禁用 future annotations）
├── serial.py        # ORM → RpcModel 转换函数
├── service.py       # 抽象服务类（禁用 future annotations）
└── impl.py          # DatabaseBacked 实现

Registration：发现包注册

服务包 MUST 是下述发现包的子包（启动时由 list_all_service_method_signatures() 通过 pkgutil.walk_packages 扫描并注册到全局注册表 _global_service_registry）。技能文档列出 12 个，当前源码中的 service_packages 元组实际为 13 个条目（多出 sentry.relocation.services）：

sentry.auth.services
sentry.audit_log.services
sentry.backup.services
sentry.hybridcloud.services
sentry.identity.services
sentry.integrations.services
sentry.issues.services
sentry.notifications.services
sentry.organizations.services
sentry.projects.services
sentry.relocation.services
sentry.sentry_apps.services
sentry.users.services

如果业务归属不在这份清单里，可以像 service.py 注释 建议的那样，向 service_packages 元组追加新条目后再进行发现。

model.py：RpcModel 定义

模型文件同时给 CELL 与 CONTROL 示例：

# Please do not use
#     from __future__ import annotations
# in modules such as this one where hybrid cloud data models or service classes are
# defined, because we want to reflect on type annotations and avoid forward references.

import datetime
from typing import Any

from pydantic import Field

from sentry.hybridcloud.rpc import DEFAULT_DATE, RpcModel


class RpcMyThing(RpcModel):
    id: int = 0
    organization_id: int = 0
    name: str = ""
    is_active: bool = True
    # Use repr=False for opaque blobs and sensitive data to prevent log leakage
    config: dict[str, Any] = Field(repr=False, default_factory=dict)
    date_added: datetime.datetime = DEFAULT_DATE

    class Config:
        orm_mode = True
        use_enum_values = True


class RpcMyThingUpdate(RpcModel):
    """Write model for updates — only include mutable fields."""
    name: str = ""
    is_active: bool = True

Control 侧模型同理，注意 token 这类凭据字段的写法：

class RpcMyMapping(RpcModel):
    id: int = 0
    user_id: int | None = None
    organization_id: int = 0
    role: str = ""
    email: str | None = None
    # Use repr=False for tokens, secrets, and credentials
    token: str = Field(repr=False, default="")
    date_added: datetime.datetime = DEFAULT_DATE

    class Config:
        orm_mode = True
        use_enum_values = True

serial.py：ORM 序列化

最简单且最推荐的写法是 serialize_by_field_name（按字段名自动映射）。字段名不同时可用 name_transform，值需要转换（例如枚举转 int）时用 value_transform，映射复杂或含计算字段时退化为手动构造：

def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(obj)

# name_transform：ORM 字段名与 RPC 字段名不一致时
def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(
        obj,
        name_transform=lambda n: f"thing_{n}" if n == "id" else n,
    )

# value_transform：值需要转换（如枚举 → int）
def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(
        obj,
        value_transform=lambda v: v.value if hasattr(v, "value") else v,
    )

# 手动构造：映射复杂时
def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing(
        id=obj.id,
        organization_id=obj.organization_id,
        name=obj.name,
        is_active=obj.flags.is_active,
        date_added=obj.date_added,
    )

service.py：抽象服务类

CELL 服务用 @cell_rpc_method(resolve=ByOrganizationId())，CONTROL 服务用 @rpc_method。装饰器必须位于 @abstractmethod 之前。文件底部必须做模块级委托实例化：

# service.py (CELL silo)
from abc import abstractmethod

from sentry.hybridcloud.rpc.resolvers import ByOrganizationId
from sentry.hybridcloud.rpc.service import RpcService, cell_rpc_method
from sentry.mydomain.services.mything.model import RpcMyThing, RpcMyThingUpdate
from sentry.silo.base import SiloMode


class MyThingService(RpcService):
    key = "my_thing"
    local_mode = SiloMode.CELL

    @classmethod
    def get_local_implementation(cls) -> RpcService:
        from sentry.mydomain.services.mything.impl import DatabaseBackedMyThingService
        return DatabaseBackedMyThingService()

    @cell_rpc_method(resolve=ByOrganizationId())
    @abstractmethod
    def get_by_id(self, *, organization_id: int, id: int) -> RpcMyThing | None:
        pass

    @cell_rpc_method(resolve=ByOrganizationId())
    @abstractmethod
    def update(self, *, organization_id: int, id: int, attrs: RpcMyThingUpdate) -> RpcMyThing | None:
        pass


my_thing_service = MyThingService.create_delegation()

# service.py (CONTROL silo)
from abc import abstractmethod

from sentry.hybridcloud.rpc.service import RpcService, rpc_method
from sentry.mydomain.services.mymapping.model import RpcMyMapping
from sentry.silo.base import SiloMode


class MyMappingService(RpcService):
    key = "my_mapping"
    local_mode = SiloMode.CONTROL

    @classmethod
    def get_local_implementation(cls) -> RpcService:
        from sentry.mydomain.services.mymapping.impl import DatabaseBackedMyMappingService
        return DatabaseBackedMyMappingService()

    @rpc_method
    @abstractmethod
    def get_by_user(self, *, user_id: int) -> list[RpcMyMapping]:
        pass

    @rpc_method
    @abstractmethod
    def upsert(self, *, organization_id: int, user_id: int | None = None, role: str = "") -> RpcMyMapping:
        pass


my_mapping_service = MyMappingService.create_delegation()

真实的仓库实例可对照 OrganizationService：key = "organization"、local_mode = SiloMode.CELL，方法上同时混用带 resolver 的 @cell_rpc_method 抽象方法与普通便捷方法。

impl.py：数据库实现

DatabaseBacked 子类必须实现每一个 @abstractmethod，且参数名必须与抽象签名完全一致（框架用 inspect.signature 做严格比对，见 service.py）。注意只有 service.py 与 model.py 禁用了 from __future__ import annotations，serial.py 与 impl.py 可以用：

# impl.py (CELL silo example)
class DatabaseBackedMyThingService(MyThingService):
    def get_by_id(self, *, organization_id: int, id: int) -> RpcMyThing | None:
        try:
            obj = MyThing.objects.get(organization_id=organization_id, id=id)
        except MyThing.DoesNotExist:
            return None
        return serialize_my_thing(obj)

    def update(self, *, organization_id: int, id: int, attrs: RpcMyThingUpdate) -> RpcMyThing | None:
        try:
            obj = MyThing.objects.get(organization_id=organization_id, id=id)
        except MyThing.DoesNotExist:
            return None
        obj.name = attrs.name
        obj.is_active = attrs.is_active
        obj.save()
        return serialize_my_thing(obj)

Control 侧实现往往需要处理并发写冲突，可参考模板中的 transaction.atomic + update_or_create + IntegrityError 回退方案。

init.py：重导出

from .model import *  # noqa
from .service import *  # noqa

新服务检查清单

  • [ ] key 在所有服务中唯一（用 grep -r 'key = "' src/sentry/*/services/*/service.py 核查）
  • [ ] local_mode 与数据实际所在位置一致
  • [ ] get_local_implementation() 返回 DatabaseBacked 子类实例
  • [ ] service.py 底部有模块级 my_service = MyService.create_delegation()
  • [ ] __init__.py 重导出模型与服务
  • [ ] service.py、model.py 无 from __future__ import annotations

第四步：添加或修改方法

CELL 服务方法

@cell_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def my_method(
    self,
    *,
    organization_id: int,
    name: str,
    options: RpcMyOptions | None = None,
) -> RpcMyResult | None:
    pass

关键规则：

  • @cell_rpc_method MUST 位于 @abstractmethod 之前；
  • resolver 所用到的参数（如 organization_id）MUST 出现在方法签名中；
  • 当返回类型是 Optional、且"缺少组织映射"应视为"未找到"而非错误时，使用 return_none_if_mapping_not_found=True（框架在解析不到 Cell 时提前短路返回 None，见 service.py）。

CONTROL 服务方法

@rpc_method
@abstractmethod
def my_method(
    self,
    *,
    user_id: int,
    data: RpcMyData,
) -> RpcMyResult:
    pass

非抽象便捷方法

可以在抽象类里加非抽象方法去组合多个 RPC 调用，它们只在本进程内执行，不对外暴露为 RPC 端点：

def get_by_slug_or_id(self, *, slug: str | None = None, id: int | None = None) -> RpcThing | None:
    if slug:
        return self.get_by_slug(slug=slug)
    if id:
        return self.get_by_id(id=id)
    return None

错误传播：只能走返回类型

RPC 方法的错误 MUST 通过返回类型表达——远端异常会被重包装为通用的 Invalid service request 再返回给外部调用方。因此常见模式是"结果信封"：

class RpcTentativeResult(RpcModel):
    success: bool
    error_str: str | None
    result: str | None

class DatabaseBackedMyService(MyService):
    def foobar(self, *, organization_id: int) -> RpcTentativeResult:
        try:
            some_function_call()
        except e:
            return RpcTentativeResult(success=False, error_str=str(e))

        return RpcTentativeResult(success=True, result="foobar")

RpcModel 设计规范

RPC 模型是 sentry.hybridcloud.rpc.RpcModel（Pydantic BaseModel 子类，定义见 rpc/init.py）的实例，它本质是 Silo 之间的序列化契约。

支持的类型与默认值

类型 默认值 说明
int 0 或 -1
str ""
bool False / True
float 0.0
int | None None
str | None None
list[T] Field(default_factory=list) T 必须可序列化
dict[str, T] Field(default_factory=dict) 键必须为 str
RpcModel 子类 嵌套模型实例
Enum 子类 枚举成员 需设置 use_enum_values = True
datetime.datetime DEFAULT_DATE 从 sentry.hybridcloud.rpc 导入

不支持（会破坏序列化）：set、frozenset、tuple、bytes、Django 模型实例、Any、前向引用字符串、来自 from __future__ import annotations 的类型。

标准模型模板

import datetime
from typing import Any

from pydantic import Field

from sentry.hybridcloud.rpc import DEFAULT_DATE, RpcModel


class RpcMyThing(RpcModel):
    id: int = -1
    name: str = ""
    tags: list[str] = Field(default_factory=list)
    config: dict[str, Any] = Field(repr=False, default_factory=dict)
    secret_token: str = Field(repr=False, default="")
    date_added: datetime.datetime = DEFAULT_DATE
    parent_id: int | None = None

    class Config:
        orm_mode = True
        use_enum_values = True

用 repr=False 隐藏敏感字段

任何包含敏感、臃肿或不透明数据的字段都应加 Field(repr=False)，防止值出现在 repr() 输出、日志、Sentry breadcrumbs 与 traceback 中。判断何时使用：

字段类型 例子 隐藏原因
密钥与凭据 tokens、哈希密码、API keys、client secrets 防止密钥泄漏进日志
不透明 blob config、metadata、extra_data 字典 日志噪声大，可能含 PII 或凭据
会话/认证材料 session nonces、OAuth tokens 安全敏感
请求/响应体 headers、payloads 可能含认证头或 PII

仓库里的真实案例（详见 rpc-models.md）：orgauthtoken/model.py 的 token_hashed、app/model.py 的 client_id/client_secret、user/model.py 的 session_nonce、lost_password_hash/model.py 的 hash、integration/model.py 的 metadata、usersocialauth/model.py 的 extra_data 等全部使用 Field(repr=False)。

经验法则：字段名包含 token、secret、key、hash、password、nonce、credential、config、metadata、extra_data、headers 任意一词，就应加 repr=False。

常见陷阱

  1. 缺少默认值：每个字段 MUST 有默认值——Pydantic 要求这样才能在后续新增字段时顺利反序列化旧版本发出的载荷；
  2. from __future__ import annotations：会把所有注解变成字符串，破坏 Pydantic 反射，模型文件一律禁用；
  3. 可变默认值：list/dict 默认值用 Field(default_factory=list)，不能写 = []；
  4. Enum 字段：字段类型声明为 int（或 str），在 Config 中设置 use_enum_values = True，默认值使用枚举成员；
  5. 嵌套 RpcModel：自动序列化，但要确保嵌套模型同样有默认值；
  6. 读模型 vs 写模型：读取响应（RpcMyThing）与写入载荷（RpcMyThingUpdate）分开定义，写模型只放可变字段；
  7. 漏写 repr=False：敏感字段会出现在日志与错误报告中，新增字段时必须审计敏感性。

Cell 解析策略：resolver 全解

CELL 服务（local_mode = SiloMode.CELL）的每个方法都要通过 @cell_rpc_method 的 resolve 参数声明"如何从方法参数解析目标 Cell"。所有 resolver 定义在 src/sentry/hybridcloud/rpc/resolvers.py。

Resolver 解析依据 默认 parameter_name 适用场景
ByOrganizationId 组织 ID → 经 OrganizationMapping 定位 Cell "organization_id" 方法含 organization_id: int 参数
ByOrganizationSlug 组织 slug → 经 OrganizationMapping 定位 Cell "slug" 方法含 slug: str 参数
ByOrganizationIdAttribute RpcModel 参数的某属性 → 组织 ID → Cell （必须显式给出） 方法含带组织 ID 字段的 RpcModel
ByCellName Cell 名字符串 → Cell "cell_name" 调用方已知道 Cell 名
RequireSingleOrganization 仅适用于单组织环境 （无） 方法只在单组织模式工作

各 resolver 用法示例

# ByOrganizationId（最常用），默认 parameter_name="organization_id"
@cell_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def get_thing(self, *, organization_id: int, id: int) -> RpcThing | None:
    pass

# 自定义参数名：这里参数叫 id 而不是 organization_id
@cell_rpc_method(resolve=ByOrganizationId("id"))
@abstractmethod
def serialize_organization(self, *, id: int) -> Any | None:
    pass

第二种写法正是 OrganizationService.serialize_organization 的真实用法。

# ByOrganizationSlug
@cell_rpc_method(resolve=ByOrganizationSlug())
@abstractmethod
def get_org_by_slug(self, *, slug: str) -> RpcOrgSummary | None:
    pass

# ByOrganizationIdAttribute：parameter_name 必填（指方法参数名），attribute_name 默认 "organization_id"
# 框架会执行 arguments["organization_member"].organization_id 取组织 ID 再查 Cell
@cell_rpc_method(resolve=ByOrganizationIdAttribute("organization_member"))
@abstractmethod
def update_membership_flags(self, *, organization_member: RpcOrganizationMember) -> None:
    pass

# 使用不同属性名
@cell_rpc_method(resolve=ByOrganizationIdAttribute("request", attribute_name="org_id"))
@abstractmethod
def process_request(self, *, request: RpcMyRequest) -> None:
    pass

# ByCellName
@cell_rpc_method(resolve=ByCellName())
@abstractmethod
def update_cell_user(self, *, user: RpcCellUser, cell_name: str) -> None:
    pass

# RequireSingleOrganization：非单组织环境下会抛 CellResolutionError
@cell_rpc_method(resolve=RequireSingleOrganization())
@abstractmethod
def get_default_organization(self) -> RpcOrganization:
    pass

return_none_if_mapping_not_found

当方法返回 Optional、且"找不到组织映射"应被当作"不存在"（而非错误）时开启：

@cell_rpc_method(resolve=ByOrganizationId("id"), return_none_if_mapping_not_found=True)
@abstractmethod
def get_organization_by_id(self, *, id: int) -> RpcOrganization | None:
    pass

不带此标志时，缺失 OrganizationMapping 会抛 CellMappingNotFound；带此标志则直接返回 None。适用场景：lookup 型方法，需优雅处理已删除或未映射的组织；禁止场景：缺失映射意味着 bug 或数据完整性问题的写路径。

选择决策树

  1. 方法直接收 organization_id: int？→ ByOrganizationId()
  2. 方法收 slug: str？→ ByOrganizationSlug()
  3. 方法收含 organization_id 的 RpcModel？→ ByOrganizationIdAttribute("param_name")
  4. 调用方已知 Cell 名？→ ByCellName()
  5. 单组织专属操作？→ RequireSingleOrganization()

签名演进：兼容与破坏性变更

安全变更（向后兼容，可随时做）

  • 新增带默认值的可选参数
  • 放宽返回类型（如 Control RPC 服务上 RpcFoo → RpcFoo | None）
  • 给 RpcModel 新增带默认值的字段

破坏性变更（需要跨版本协调）

  • 删除或重命名参数
  • 修改参数类型
  • 收窄返回类型
  • 从 RpcModel 删除字段

破坏性变更采用两阶段策略：

  1. 新增方法，旧方法暂时保留；
  2. 把所有调用方迁移到新方法；
  3. 迁移完成后按弃用流程删除旧方法。

弃用与删除：三阶段安全下线

跨 Silo 部署窗口内各 Silo 可能运行不同代码版本，因此移除 RPC 方法必须走三阶段流程（完整版见 deprecation.md）：运行时禁用 → 迁移调用方 → 删除代码。

Phase 1：运行时禁用

利用 hybrid_cloud.rpc.disabled-service-methods 选项把方法禁用掉而不删任何代码，保证可即时回滚。该方法在远端被调用时会直接抛 RpcDisabledException，不会发起网络请求——对应源码中的 _RemoteSiloCall._check_disabled()（见 service.py）。

options.set("hybrid_cloud.rpc.disabled-service-methods", [
    "MyService.old_method",
])

运行时控制选项：

选项键 类型 作用
hybrid_cloud.rpc.disabled-service-methods list[str] 禁用指定方法（格式 "ServiceName.method_name"）
hybridcloud.rpc.retries int 全局重试次数（默认 5）
hybridcloud.rpc.method_retry_overrides dict[str, int] 按方法覆盖重试次数（键："service_key.method_name"）
hybridcloud.rpc.method_timeout_overrides dict[str, float] 按方法覆盖超时（键："service_key.method_name"）

技巧：删除方法前先把其重试设为 0、超时调到很低，让残留调用方立刻暴露出来。

Phase 2：迁移调用方

  1. 全仓搜索调用方：grep -r "my_service\.old_method" src/ tests/
  2. 逐一改为新方法或内联逻辑；
  3. 过渡期保留实现，但加弃用注释：
    @cell_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def old_method(self, *, organization_id: int) -> RpcThing | None:
    """Deprecated: Use new_method instead. Remove after YYYY-MM-DD."""
    pass
  4. 部署并通过指标确认该方法零流量：
    • 检查 hybrid_cloud.dispatch_rpc.response_code 中 rpc_method=ServiceName.old_method 标签；
    • 确认 Sentry 无 RpcDisabledException 报错。

Phase 3：删除代码

  1. 从 service.py 删除抽象方法；
  2. 从 impl.py 删除实现；
  3. 删除 model.py 中仅被该方法使用的 RpcModel；
  4. 清理 serial.py 中不再使用的序列化辅助函数；
  5. 从 hybrid_cloud.rpc.disabled-service-methods 移除该条目；
  6. 删除相关测试。

删除整个服务时，先对每个方法完成三阶段，然后删除 create_delegation() 调用、服务类与整个服务目录；由于 list_all_service_method_signatures() 通过 pkgutil.walk_packages 动态发现，服务会自然从注册表消失。

安全删除检查清单：生产环境已通过 disabled 选项禁用 ✓；指标确认零流量至少 1 周 ✓；grep 确认无调用方 ✓；测试已更新/删除 ✓；无用 RpcModel 已清理 ✓；代码删除后已移除 disabled 列表条目 ✓。

测试：三类必测 + 全套工具

每个 RPC 服务都需要三类测试：Silo 模式兼容性、数据准确性、错误处理。当测试依赖 outbox 处理或 on_commit 钩子时，用 TransactionTestCase 而非 TestCase。

7.1 用 @all_silo_test 覆盖全部模式

每个服务测试类 MUST 使用 @all_silo_test，让测试在 MONOLITH、CELL、CONTROL 三种模式各跑一遍，确保委托层在本地与远端两条派发路径上都工作：

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, create_test_cells

@all_silo_test
class MyServiceTest(TestCase):
    def test_get_by_id(self):
        org = self.create_organization()
        result = my_service.get_by_id(organization_id=org.id, id=thing.id)
        assert result is not None

需要命名 Cell（例如验证 Cell 解析）时：

@all_silo_test(cells=create_test_cells("us", "eu"))
class MyServiceCellTest(TransactionTestCase):
    ...

在单测内访问其他 Silo 的 ORM 模型时，用 assume_test_silo_mode / assume_test_silo_mode_of 临时切换模式：

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.CELL):
        org = self.create_organization()
    result = my_service.get_by_id(organization_id=org.id, id=thing.id)
    assert result is not None

7.2 用 dispatch_to_local_service 验证序列化往返

测试参数与返回值经序列化/反序列化后仍完好：

from sentry.hybridcloud.rpc.service import dispatch_to_local_service

def test_serialization_round_trip(self):
    result = dispatch_to_local_service(
        "my_service_key",
        "my_method",
        {"organization_id": org.id, "name": "test"},
    )
    assert result["value"] is not None

7.3 RPC 模型数据准确性

把 RPC 模型的每个字段与源 ORM 对象逐项比对；带 flags 或嵌套对象时遍历全部字段名；list 结果先按 ID 排序再比对：

def test_rpc_model_accuracy(self):
    orm_obj = MyModel.objects.get(id=thing.id)
    rpc_obj = my_service.get_by_id(organization_id=org.id, id=thing.id)
    assert rpc_obj.id == orm_obj.id
    assert rpc_obj.name == orm_obj.name
    assert rpc_obj.organization_id == orm_obj.organization_id
    assert rpc_obj.is_active == orm_obj.is_active
    assert rpc_obj.date_added == orm_obj.date_added

def test_flags_accuracy(self):
    rpc_org = organization_service.get(id=org.id)
    for field_name in rpc_org.flags.get_field_names():
        assert getattr(rpc_org.flags, field_name) == getattr(orm_org.flags, field_name)

def test_list_accuracy(self):
    rpc_items = my_service.list_things(organization_id=org.id)
    orm_items = list(MyModel.objects.filter(organization_id=org.id).order_by("id"))
    assert len(rpc_items) == len(orm_items)
    for rpc_item, orm_item in zip(sorted(rpc_items, key=lambda x: x.id), orm_items):
        assert rpc_item.id == orm_item.id
        assert rpc_item.name == orm_item.name

7.4 跨 Silo 资源创建

若服务会经 outbox 或 mapping 创建跨 Silo 传播的资源，需验证跨 Silo 效果。outbox_runner() 可在测试中同步 flush outbox；HybridCloudTestMixin 提供常用跨 Silo 断言：

from sentry.testutils.outbox import outbox_runner

def test_cross_silo_mapping_created(self):
    with outbox_runner():
        my_service.create_thing(organization_id=org.id, name="test")
    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
        assert mapping.name == "test"

三重相等断言（RPC 结果 = 源 ORM = 跨 Silo 副本）：

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.CELL):
        orm_obj = MyModel.objects.get(id=rpc_result.id)
    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
    assert rpc_result.slug == orm_obj.slug == mapping.slug

使用通用断言混合类：

from sentry.testutils.hybrid_cloud import HybridCloudTestMixin

class MyServiceTest(HybridCloudTestMixin, TransactionTestCase):
    def test_member_mapping_synced(self):
        self.assert_org_member_mapping(org_member=org_member)

7.5 错误处理

覆盖所有 Silo 模式下的错误路径：未找到返回 None、禁用方法抛 RpcDisabledException、远端异常被包装成 RpcRemoteException、失败操作无副作用。所有直接/间接调用代码也应使用正确的 Silo 装饰器进行测试。

def test_not_found_returns_none(self):
    result = my_service.get_by_id(organization_id=org.id, id=99999)
    assert result is None

def test_missing_org_returns_none(self):
    # 适用于 return_none_if_mapping_not_found=True 的方法
    result = my_service.get_by_id(organization_id=99999, id=1)
    assert result is None

def test_disabled_method_raises(self):
    with override_options({"hybrid_cloud.rpc.disabled-service-methods": ["MyService.my_method"]}):
        with pytest.raises(RpcDisabledException):
            dispatch_remote_call(None, "my_service_key", "my_method", {"id": 1})

def test_remote_error_wrapping(self):
    if SiloMode.get_current_mode() == SiloMode.CELL:
        with pytest.raises(RpcRemoteException):
            my_control_service.do_thing_that_fails(...)

def test_no_side_effects_on_failure(self):
    result = my_service.create_conflicting_thing(organization_id=org.id)
    assert not result
    with assume_test_silo_mode(SiloMode.CELL):
        assert not MyModel.objects.filter(organization_id=org.id).exists()

7.6 测试关键导入汇总

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    cell_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.hybridcloud.rpc.service import (
    dispatch_to_local_service,
    dispatch_remote_call,
    RpcDisabledException,
    RpcRemoteException,
)

仓库真实测试可参考 tests/sentry/hybridcloud/ 下的 test_rpc.py、test_organization.py 等服务测试与 services/ 下的服务级测试，其组织方式与本技能描述的用例分类一致。

提交前的最终自检

  • [ ] service.py / model.py 无 from __future__ import annotations
  • [ ] 所有 RPC 参数 keyword-only（* 分隔符）
  • [ ] 所有参数有显式类型注解
  • [ ] 所有类型可序列化（基础类型、RpcModel、list、Optional、dict、Enum、datetime）
  • [ ] Cell 服务方法带 @cell_rpc_method 且配置了合适的 resolver
  • [ ] Control 服务方法带 @rpc_method
  • [ ] @cell_rpc_method / @rpc_method 位于 @abstractmethod 之前
  • [ ] service.py 底部模块级调用了 create_delegation()
  • [ ] 服务包位于发现包清单内
  • [ ] impl.py 实现了每个抽象方法且参数名一致
  • [ ] serial.py 正确完成 ORM → RPC 模型转换
  • [ ] 敏感字段使用 Field(repr=False)（token、secret、config、metadata 等）
  • [ ] 测试使用 @all_silo_test 覆盖全部 Silo 模式
  • [ ] 测试校验 RPC 模型与 ORM 对象的字段准确性
  • [ ] 测试校验跨 Silo 资源（mapping、replica）以正确数据创建
  • [ ] 测试覆盖错误场景（not found、禁用方法、失败操作）
  • [ ] 测试经 dispatch_to_local_service 覆盖序列化往返

结语与延伸阅读

混合云 RPC 是 Sentry 演进到多 Cell 架构的枢纽层。技能文档的价值在于把"接口声明即契约、注解反射驱动序列化、Cell 归属决定路由、删除必须分级"这四件事固化成了一套可审查的工程规范。想要进一步深入，可以沿着 service.py 阅读 DelegatingRpcService 与 _RemoteSiloCall 的完整派发实现，对照 rpc/init.py 中的 RpcModel、DEFAULT_DATE 与 silo_mode_delegation 工具，再结合 cell-architecture 技能文档（同仓库 .agents/skills 目录）理解 Cell 拓扑本身；改动完成后务必在三种 Silo 模式下跑通测试，让委托层与远端派发两条路径都得到验证。

sentry
Developer-first error tracking and performance monitoring
项目地址：https://gitcode.com/GitHub_Trending/sen/sentry
登录后查看全文

相关内容推荐

1 Sentry Hybrid Cloud RPC 方法下线与删除的三阶段安全指南：Deprecating & Removing RPC Methods2 Sentry 混合云 RPC 服务测试参考指南：silo 兼容、序列化往返与跨 silo 效应验证3 Sentry 混合云 RPC 服务文件模板深度解析：从零搭建 model / serial / service / impl 四件套4 Sentry Django ORM 模型设计实战指南：Silo 归属、复制、Relocation 与跨 Silo 外键的架构决策5 Sentry Hybrid Cloud RPC 模型指南：跨 Silo 序列化契约的建模规范、字段类型与陷阱6 Sentry 混合云架构下的 Endpoint Silo 测试：从装饰器映射到跨 Silo 数据验证的完整参考7 Sentry Cell 架构全解：Cell 与 Locality 的层次、请求路由三条路径与 region 到 cell 的迁移实战指南8 Sentry 混合云（Hybrid Cloud）本地开发完全指南：Control Silo 与 Region Silo 的搭建、运行与调试9 Sentry Hybrid Cloud Outbox Signal Receiver 开发指南：Cell 与 Control Silo 手动接收器的编写与落盘逻辑10 Sentry Cell Resolution（Cell 路由解析）实战指南：读懂 cell_rpc_method 的五大 Resolver 策略
热门项目推荐
相关项目推荐

热门内容推荐

1 编程实践项目探索指南：从零构建技术能力体系2 技术解构式学习：从0到1构建你的编程知识体系3 构建自己的技术世界：build-your-own-x项目的实践探索指南4 解锁编程技能的实践之旅：从零构建你的技术世界5 技术实践探索：从零开始构建核心系统的实践指南6 亲手锻造技术引擎：从0到1构建核心系统的实践指南

最新内容推荐

daisyUI FAB 组件完整实战指南：悬浮按钮、Speed Dial 速拨菜单与 fab-flower 扇形展开three.js TAA 抗锯齿实战：TAARenderPass 时序累积采样后处理全解析OpenHuman Subconscious Factory 重构（Phase 5）落地指南：测试矩阵、文档同步与多实例发布策略curl 的 `--dns-interface` 参数全解析：让 DNS 查询从指定网络接口发出freeCodeCamp 每日编程挑战解析：用 JavaScript 实现 SCREAMING_SNAKE_CASE 变量名转换器Langchain-Chatchat 配置体系实战指南：Settings 统一管理与 YAML 配置模板全解析Langchain-Chatchat 会话模型深度解析：ConversationModel 的表结构、生命周期与多端调用链路Arbor 执行器任务书（Executor Brief）编写指南：用单假设实验契约驱动 scientific-agent-skills 中的 HTR 自主优化three.js 后处理进阶：使用 TexturePass 将纹理/渲染结果铺满整个屏幕ToolJet Marketplace 插件开发实战：使用 tooljet CLI 从零构建 GitHub 数据源插件

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.14 K
2.75 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
857
1.35 K
docsdocs
暂无描述
Markdown
898
5.82 K
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
921
1.84 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.8 K
1.02 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
531
596
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
1.02 K
519
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库，实现网络在NPU上加速计算。
C++
1.36 K
1.46 K
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR（Multi-Level Intermediate Representation）构建的，面向昇腾亲和算子编译时使用的中间表示，提供昇腾完备表达能力，通过编译优化提升昇腾AI处理器计算效率，支持通过生态框架使能昇腾AI处理器与深度调优
C++
548
391