Sentry 通知平台如何注册一个新的 Notification Action 并实现 fire 逻辑

Sentry 通知平台如何注册一个新的 Notification Action 并实现 fire 逻辑

【免费下载链接】sentryDeveloper-first error tracking and performance monitoring项目地址: https://gitcode.com/GitHub_Trending/sen/sentry

当你在 Sentry 中要接入一种新的通知行为——比如把审计日志推送到 Slack 频道，或为某种触发事件对接第三方服务——需要为NotificationAction模型注册一个ActionRegistration子类：用@NotificationAction.register_action声明 trigger/service/target 组合，并在子类中实现fire()方法承载与下游服务的实际通信逻辑。本文基于仓库内 notification_actions.md 的设计说明、notificationaction.py 的模型实现，以及 test_notificationaction.py 中的验证用例，给出从注册到验证 fire 逻辑的完整路径。

Notification Action 由什么组成

根据 notification_actions.md，Notification Action 依赖四个部分：

1. Triggers——通知的来源，即 Sentry 中发生了什么导致通知触发；
2. Services——投递机制（Slack、PagerDuty、MSTeams、Sentry Notifications 等）；
3. Targets——接收方类型（用户、团队，还是集成特有目标）；
4. Registrations——帮助 setup 新 action 的ActionRegistration子类。

注册键由三元组唯一确定，notificationaction.py 中get_registry_key的格式是"{trigger_type}:{service_type}:{target_type}"。同一组合只允许一个注册类。

前提：确认三个枚举中已存在你要用的值

register_action在注册时会对三个类型做校验（notificationaction.py）：trigger_type必须在ActionTrigger.as_choices()中、service_type必须在ActionService.as_choices()中、target_type必须在ActionTarget.as_choices()中，否则直接抛出AttributeError，例如：

Trigger type of {trigger_type} is not registered. Modify ActionTrigger.

文档明确说明：这些枚举在保存新的 Notification Action 之前会被 Django 用于校验，所以如果需要新的 trigger/service/target，先扩展 notificationaction.py 中对应的枚举，这一步不能跳过。

当前枚举中已有的值供选择：

• ActionService：EMAIL=0、PAGERDUTY=1、SLACK=2、MSTEAMS=3、SENTRY_APP=4、SENTRY_NOTIFICATION=5、OPSGENIE=6、DISCORD=7、SLACK_STAGING=8；
• ActionTarget：SPECIFIC=0、USER=1、TEAM=2、SENTRY_APP=3、ISSUE_OWNERS=4；
• ActionTrigger：AUDIT_LOG=0（audit-log）、GS_SPIKE_PROTECTION=100（spike-protection）。

如果目标组合的三个值都已存在（比如AUDIT_LOG+SENTRY_NOTIFICATION+SPECIFIC），可以直接进入下一步。

编写注册类并实现 fire()

ActionRegistration基类定义在 notificationaction.py：

• fire(self, data)是抽象方法，每个注册类必须实现，文档说明这里就是“与目标服务通信的逻辑”所在；
• validate_action(data)在通过 API 校验新 action 时被调用，用于数据库完整性检查之外的自定义校验，不需要时可保留默认实现；
• serialize_available(organization, integrations)用于把该 action 的可用性序列化给前端，默认返回空列表[]。

另外注意ActionRegistration.__init__会保存self.action（触发本次 fire 的NotificationAction实例），模型自身的fire()通过registration(action=self).fire(*args, **kwargs)调用你的实现，因此你可以在fire()内通过self.action读取trigger_type、service_type（即type字段）、target_identifier、target_display等配置。

按照 notification_actions.md 给出的示例，注册写法如下（fire内部按你的服务对接逻辑填充）：

from typing import Any from sentry.notifications.models.notificationaction import ( ActionRegistration, ActionService, ActionTarget, ActionTrigger, NotificationAction, ) @NotificationAction.register_action( trigger_type=ActionTrigger.AUDIT_LOG.value, service_type=ActionService.SENTRY_NOTIFICATION.value, target_type=ActionTarget.SPECIFIC.value, ) class SentryAuditLogRegistration(ActionRegistration): def fire(self, data: Any) -> None: # 与目标服务通信的逻辑写在这里 ... @classmethod def validate_action(cls, data) -> None: pass @classmethod def serialize_available(cls, organization, integrations=None) -> list[Any]: return []

两个注册时会立即失败的约束（来自 register_action 的源码）：

• 三个类型值任一不在对应枚举的as_choices()中，抛AttributeError，提示去修改相应枚举；
• 组合键"{trigger}:{service}:{target}"已存在于_registry中时，抛AttributeError：Existing registration found for trigger:..., service:..., target:....

装饰器在类定义被 import 时执行注册，因此定义注册类的模块必须在应用启动时被导入，否则运行时_registry中查不到该组合。

验证注册是否生效、fire 是否被调用

仓库中的 test_notificationaction.py 给出了可复用的验证模式，核心是用@patch.dict(NotificationAction._registry, {})清空注册表后注册一个 handler，再触发fire()：

from unittest.mock import MagicMock, patch from sentry.notifications.models.notificationaction import ( NotificationAction, ) from sentry.notifications.models.notificationaction import logger as NotificationActionLogger from sentry.testutils.cases import TestCase @patch.dict(NotificationAction._registry, {}) class NotificationActionTest(TestCase): def setUp(self) -> None: self.organization = self.create_organization(name="night city") self.projects = [ self.create_project(name="netrunner", organization=self.organization), self.create_project(name="edgerunner", organization=self.organization), ] self.notif_action = self.create_notification_action( organization=self.organization, projects=self.projects ) @patch.object(NotificationActionLogger, "error") def test_register_action_for_fire(self, mock_error_logger: MagicMock) -> None: mock_handler = MagicMock() NotificationAction.register_action( trigger_type=self.notif_action.trigger_type, service_type=self.notif_action.service_type, target_type=self.notif_action.target_type, )(mock_handler) self.notif_action.fire() assert not mock_error_logger.called assert mock_handler.called

成功条件是两条断言：mock_handler.called（fire()被正确路由到注册的 handler）且mock_error_logger未被调用（没有走到错误分支）。

同一文件还覆盖了两条失败路径，写新注册类时都应注意：

• 重复注册：对同一组合第二次调用register_action会抛AttributeError（见test_register_action_for_overlap）；
• 组合未注册：fire()不会抛异常，而是静默失败，只记录一条missing_registration错误日志（见test_fire_fails_silently和 NotificationAction.fire 的实现）。反之，注册命中时记录的是fire_action日志，其中包含action_id、trigger、service、target四个字段，可用于线上核对。

如果你的 action 还要通过 API 创建，notification_action_request.py 中的validate_with_registry会再查一次注册表：组合未注册时返回错误Combination of trigger_type, service_type and target_type has not been registered.，随后调用你注册类上的validate_action(data)。另外当service_type是PAGERDUTY、SLACK、SLACK_STAGING、MSTEAMS、OPSGENIE之一时，API 层还要求提供该组织下已安装的对应integration_id，否则保存直接被拒绝。

限制与注意事项

• fire()是抽象方法，不能省略；validate_action和serialize_available有默认实现，可按需覆盖。若需要前端在可用 action 列表里看到你的组合，必须实现serialize_available，默认返回空列表意味着不会出现在前端。
• 组合未注册时fire()不抛异常、只记错误日志，排查“通知没发出去”时应先查missing_registration日志，确认触发时该组合是否已完成注册。
• 枚举扩展是保存 Notification Action 的硬前提，只写注册类而不动枚举，在需要新触发类型时无法通过校验。

完成上述步骤后，你的注册类会进入NotificationAction._registry，模型fire()能路由到它，API 校验也能接受对应组合的 action 创建请求——这三个验证点分别对应运行时投递、API 保存和前端可见性，覆盖一个新 Notification Action 从注册到 fire 的完整链路。

【免费下载链接】sentryDeveloper-first error tracking and performance monitoring项目地址: https://gitcode.com/GitHub_Trending/sen/sentry