hybrid-cloud-rpc

ハイブリッドクラウド RPC サービス

このスキルは、Sentry のハイブリッドクラウドアーキテクチャにおける RPC サービスの作成、変更、非推奨化をガイドします。RPC サービスは、Control サイロ（ユーザー認証、組織管理）と Cell サイロ（プロジェクトデータ、イベント、イシュー、請求）の間のクロスサイロ通信を可能にします。

重要な制約

service.py または model.py ファイルで from __future__ import annotations を 絶対に 使用しないでください。 RPC フレームワークはインポート時に型アノテーションをリフレクションで解析します。前方参照はシリアライゼーションをサイレントに破壊します。

すべての RPC メソッドパラメータはキーワード専用でなければなりません（シグネチャで * を使用します）。

すべての パラメータと戻り値の型には完全な型アノテーションが必要です — 文字列による前方参照は不可です。

使用できるのは シリアライズ可能な型のみです: int、str、bool、float、None、Optional[T]、list[T]、dict[str, T]、RpcModel のサブクラス、Enum のサブクラス、datetime.datetime。

サービスは、登録済みの 12 個のディスカバリーパッケージのいずれかに 必ず 配置する必要があります（ステップ 3 を参照）。

機密フィールド（トークン、シークレット、キー、設定ブロブ、 メタデータ辞書）には Field(repr=False) を使用して、ログやエラーレポートへの漏洩を防いでください。 完全なガイドについては references/rpc-models.md を参照してください。

ステップ 1: 操作の決定

開発者が必要としていることを分類します:

目的	参照先
まったく新しい RPC サービスを作成する	ステップ 2、次にステップ 3
既存のサービスにメソッドを追加する	ステップ 2、次にステップ 4
既存メソッドのシグネチャを更新する	ステップ 5
メソッドまたはサービスを非推奨化・削除する	ステップ 6

ステップ 2: サイロモードの決定

サービスの local_mode は、データベースに裏付けられた実装がどこで実行されるかを決定します:

データの所在	local_mode	メソッドのデコレーター	例
Cell サイロ（プロジェクト、イベント、イシュー、組織データ、請求）	SiloMode.CELL	@cell_rpc_method(resolve=...)	OrganizationService
Control サイロ（ユーザー、認証、組織マッピング）	SiloMode.CONTROL	@rpc_method	OrganizationMemberMappingService

判断ルール: クエリする必要がある Django モデルが cell データベースにある場合は、SiloMode.CELL を使用します。control データベースにある場合は、SiloMode.CONTROL を使用します。

Cell サイロのサービスでは、リモート呼び出しをどのセルにルーティングするかをフレームワークが把握できるよう、すべての RPC メソッドに CellResolutionStrategy が必要です。完全なリゾルバー一覧については references/resolvers.md をロードしてください。

ステップ 3: 新しいサービスの作成

コピー＆ペースト用のファイルテンプレートについては references/service-template.md をロードしてください。

ディレクトリ構造

src/sentry/{domain}/services/{service_name}/
├── __init__.py      # Re-exports model and service
├── model.py         # RpcModel subclasses (NO future annotations)
├── serial.py        # ORM → RpcModel conversion functions
├── service.py       # Abstract service class (NO future annotations)
└── impl.py          # DatabaseBacked implementation

登録

サービスパッケージは、以下の 12 個の登録済みディスカバリーパッケージのいずれかのサブパッケージで なければなりません:

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.sentry_apps.services
sentry.users.services

サービスがこれらのいずれにも当てはまらない場合は、src/sentry/hybridcloud/rpc/service.py:list_all_service_method_signatures() の service_packages タプルに新しいエントリを追加してください。

新しいサービスのチェックリスト

• 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 がない

ステップ 4: メソッドの追加または更新

CELL サイロサービス向け

リゾルバーの詳細については、references/resolvers.md を参照してください。

@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 は @abstractmethod より前に置く必要があります
• リゾルバーのパラメーター（例: organization_id）はメソッドシグネチャに含める必要があります
• 戻り値の型が Optional であり、組織のマッピングが存在しないことがエラーではなく「見つからない」ことを意味する場合は、return_none_if_mapping_not_found=True を使用してください

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

impl.py での実装

DatabaseBacked のサブクラスは、まったく同じパラメーター名で、すべての @abstractmethod を実装する必要があります:

class DatabaseBackedMyService(MyService):
    def my_method(self, *, organization_id: int, name: str, options: RpcMyOptions | None = None) -> RpcMyResult | None:
        # ORM queries here
        obj = MyModel.objects.filter(organization_id=organization_id, name=name).first()
        if obj is None:
            return None
        return serialize_my_model(obj)

エラーの伝播

RPC メソッドが伝播するすべてのエラーは、戻り値の型を介して行う必要があります。エラーは 再ラップされ、外部の呼び出し元には汎用的な 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")

RPC モデル

サポートされる型、デフォルト値、シリアル化パターンについては、references/rpc-models.md を参照してください。

ステップ 5: メソッドシグネチャの更新

安全な変更（後方互換）

• デフォルト値を持つ新しい 省略可能な パラメーターの追加
• Control RPC サービスにおける戻り値の型の拡張（例: RpcFoo → RpcFoo | None）
• RpcModel へのデフォルト値を持つフィールドの追加

破壊的変更（調整が必要）

• パラメーターの削除または名前変更
• パラメーターの型の変更
• 戻り値の型の縮小
• RpcModel からのフィールドの削除

破壊的変更には、2 段階のアプローチを使用してください:

1. 古いメソッドと並行して新しいメソッドを追加する
2. すべての呼び出し元を新しいメソッドに移行する
3. 古いメソッドを削除する（ステップ 6 を参照）

ステップ 6: 非推奨化または削除

3 段階のワークフロー全体については、references/deprecation.md を参照してください。

簡単なまとめ: 実行時に無効化 → 呼び出し元を移行 → コードを削除。

ステップ 7: テスト

すべての RPC サービスには、3 つのカテゴリのテストが必要です: サイロモードの互換性、データの正確性、エラーハンドリング。テストでアウトボックス処理や on_commit フックが必要な場合は、TestCase ではなく TransactionTestCase を使用してください。

7.1 @all_silo_test によるサイロモードの互換性

すべてのサービステストクラスは @all_silo_test を使用する必要があります。これにより、テストが 3 つのモード（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

名前付きセルが必要なテスト（例: セル解決のテスト）の場合:

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

別のサイロに存在する 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 データを忠実に表現していることを検証します。RPC モデルの すべてのフィールド をソースの ORM オブジェクトと比較してください:

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)

リスト結果の場合は、比較する前に両側を ID で並べ替えます:

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 サイロ間のリソース作成

サービスがサイロ間で伝播するリソース（アウトボックスやマッピングを介して）を作成または更新する場合は、サイロ間の影響を検証してください。

テスト中にアウトボックスを同期的にフラッシュするには、outbox_runner() を使用します:

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 = サイロ間レプリカ）の場合:

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

一般的なサイロ間アサーションには HybridCloudTestMixin を使用します:

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 エラーハンドリング

サービスがすべてのサイロモードでエラーを正しく処理することをテストします:

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):
    # For methods with return_none_if_mapping_not_found=True
    result = my_service.get_by_id(organization_id=99999, id=1)
    assert result is None

無効化されたメソッドをテストします:

from sentry.hybridcloud.rpc.service import RpcDisabledException
from sentry.testutils.helpers.options import override_options

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})

リモートの例外が適切にラップされることをテストします:

from sentry.hybridcloud.rpc.service import RpcRemoteException

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,
)

ステップ 8: 検証（事前チェックリスト）

PR を送信する前に、以下を検証してください:

• service.py または model.py に from __future__ import annotations がない
• すべての RPC メソッドのパラメータがキーワード専用である（* セパレータ）
• すべてのパラメータに明示的な型アノテーションがある
• すべての型がシリアライズ可能である（プリミティブ、RpcModel、list、Optional、dict、Enum、datetime）
• Cell サービスのメソッドに適切なリゾルバを指定した @cell_rpc_method がある
• Control サービスのメソッドに @rpc_method がある
• @cell_rpc_method / @rpc_method が @abstractmethod より前に記述されている
• create_delegation() が service.py の末尾でモジュールレベルで呼び出されている
• サービスパッケージが登録済みの 12 個のディスカバリパッケージのいずれかに含まれている
• impl.py がすべての抽象メソッドを一致するパラメータ名で実装している
• serial.py が ORM モデルを RPC モデルに正しく変換している
• 機密フィールドに Field(repr=False) が使用されている（トークン、シークレット、設定、メタデータ）
• テストが完全なサイロモードのカバレッジのために @all_silo_test を使用している
• テストが ORM オブジェクトに対する RPC モデルのフィールドの正確性を検証している
• テストがクロスサイロリソース（マッピング、レプリカ）が正しいデータで作成されることを検証している
• テストがエラーケース（見つからない、無効化されたメソッド、失敗した操作）をカバーしている
• テストが dispatch_to_local_service を介したシリアライゼーションのラウンドトリップをカバーしている