Kafka 主题和模式注册表用于 Sentry

包含 Sentry 服务使用的 Kafka 主题和模式定义。

定义模式

目前仅支持 jsonschema。jsonschema 应直接放置在 schemas 目录中，然后通过 resource 属性从相关主题引用。

我们使用 jsonschema 为 JSON-和 msgpack-based 主题，因为大多数 msgpack 类型都有一个 JSON 等效类型。对于 bytestrings，我们使用 {"description": "msgpack bytes"} 来类型化，这目前只是像 {}（允许所有类型）一样被解释。

如果您不想手动编写它，我们可以通过 https://github.com/quicktype/quicktype 生成从有效载荷的初始 json schema

我的模式应该有多严格？

如果有疑问，我们建议模式仅严格到 Sentry 所需的最小程度。但是，最终是否在特定场景中使用更严格的模式由模式的拥有者决定。

添加示例消息

示例消息可以放置在 examples 目录中，并从相关主题/版本引用。

示例消息必须去除 所有 客户相关数据。这还包括组织 ID 和项目 ID 等信息，应替换为类似 project_id: 1 或 org_id: 1 的内容。

定义主题

每个主题都是 topics 目录中的一个 yaml 文件。此主题名称是一个“逻辑”主题名称，因为 Sentry 中的许多服务都支持根据需要将默认名称覆盖为不同的物理主题名称。主题名称在 Sentry 中必须是唯一的：同一名称不能用于不同类型的数据。

主题的 yaml 文件具有以下键

1. schemas. Schemas 是一个数组。以下为每个模式应提供的内容

   • version：递增整数。应从1开始。
   • compatibility_mode：none 或 backward。
   • type：可以是 json 或 msgpack。在两种情况下，我们使用 jsonschema 来定义消息架构。
   • resource：应与 schemas 目录中的文件名匹配
   • examples：应与 examples 目录中的文件名匹配

2. topic_configuration_config。用于创建主题的配置

3. services。哪些 Sentry 服务向主题生产和消费。

4. 描述.

5. 流水线.

使用架构（在 Python 中）

from sentry_kafka_schemas import get_codec, ValidationError
from sentry_kafka_schemas.schema_types.ingest_metrics_v1 import IngestMetric

SCHEMA: Codec[IngestMetric] = get_codec("ingest-metrics")

try:
    decoded = SCHEMA.decode(b'{"type": "c", ...}')
except ValidationError:
    return

# ingest-metrics schema defines retention_days as required type, so this is
# safe.
retention_days = decoded["retention_days"]

使用 Python 类型

Python 类型将在 sentry_kafka_schemas.schema_types 下自动生成。主题 foo-bar 版本 1 的架构已导出到 sentry_kafka_schemas.schema_types.foo_bar_v1。

请使用 JSON 架构和各个定义上的 title 属性来为它们分配稳定的名称。

例如

// a schema referenced from `topics/events.yaml, containing topic: events
{
    "title": "main_schema",
    "description": "Some additional information about the schema."
    "properties": {
        "subfield": {"$ref": "#/definitions/SubSchema"}
    },
    "definitions": {
        "SubSchema": {
            "type": "object",
            "title": "sub_schema"
        }
    }
}

产生

# file: sentry_kafka_schemas/schema_types/events_v1.py

class MainSchema(TypedDict, total=False):
    """Some additional information about the schema."""

    subfield: SubSchema

class SubSchema(TypedDict, total=False):
    ...

title 可以添加在任何级别，而不仅仅是 definitions 内部，以产生类型。巧妙地使用这种力量！

使用 Rust 类型

我们使用一个完全不同的库来生成 Rust 类型，因此 Rust 类型名称生成的规则也不同。Rust 类型正在开发中。

目前，需要将模式文件显式添加到 rust/build.rs 中。生成的类型可以通过 make view-rust-types、cargo doc --open 或在线在 https://docs.rs/sentry-kafka-schemas 上查看。

发布流程和开发安装

要从主分支发布新的稳定版本，请转到 Actions 并触发 Release 工作流程的新作业。

我们通常只是递增模式更改的 patch 号。例如，如果最后一个版本是 0.1.11，则下一个版本应该是 0.1.12。请检查 https://github.com/getsentry/sentry-kafka-schemas/releases 上的最新发布号。

发布新版本后，您应立即将 Sentry、Snuba 和 Relay 升级，以确保所有服务尽快同步到新架构。

您很可能会在 Snuba 或 Sentry 上工作，并已经在 PR 中使用这些类型。您可以通过在此存储库中运行 make build 并运行 pip install -e ~/projects/sentry-kafka-schemas/ 来完成此操作。

您需要重新运行 make build 来更新类型 -- 即使您以开发模式安装此包，类型也不会自动更改。

要停止在您正在工作的任何服务中使用此存储库的开发版本，您可以重新安装该存储库中的 Python 依赖项。最可能的命令是 make install-py-dev。

架构所有权

所有主题的定义、模式以及示例应有一个明确的负责人或多个共享负责人。每当添加新的模式和主题时，应更新CODEOWNERS文件中的此信息。

只需要来自一个团队/负责人的审核，而不是所有团队成员。