Sentry 消息转发

Sentry 是一个十分好用的错误监控系统，它可以实时的捕捉各类应用程序里的错误。例如 Django、Flask 以及其他的 WSGI、ASGI 应用。

但腾讯特色就是不参与这些开源的玩意，所以企业微信通知是没有内嵌的。有几个个人开发的企业微信插件，最终都是停止维护。于是只能自己整了。

为什么用 Serverless

做一个内嵌进 Sentry 的插件比较麻烦，所以选择使用 WebHook 接收信息，然后转发出去。

但为了这个消息转发做一个 Web 服务不值得（得考虑服务是否存活等等情况），所以选择函数计算 + HTTP 触发器，反正阿里云的运维比我运维要靠谱。

实践

代码

直接用 bottle 作为 WSGI 框架来接受 HTTP 触发器的请求。

import os
import json
import hmac
from hashlib import sha256

import requests
from bottle import Bottle, request, route, HTTPError

app = Bottle()
SENTRY_TOKEN = os.environ["SENTRY_TOKEN"]
SENTRY_URL = os.environ["SENTRY_URL"].strip("/")
WECHAT_BOT_URL = os.environ["WECHAT_BOT_URL"]


def send_message(text: str) -> None:
    """
    发送 markdown 格式的消息

    https://work.weixin.qq.com/help?doc_id=13376
    """
    requests.post(
        WECHAT_BOT_URL, json={"msgtype": "markdown", "markdown": {"content": text}}
    )


# 参考 sentry 文档, 解析信息
# https://docs.sentry.io/workflow/integrations/integration-platform/webhooks/?platform=python


def installation(data: dict) -> None:
    username = data["actor"]["name"]
    organization = data["data"]["installtion"]["organization"]["slug"]
    app = data["data"]["installtion"]["app"]["slug"]

    send_message(f"{username} 接入了在 {organization} 的 {app}")


def uninstallation(data: dict) -> None:
    username = data["actor"]["name"]
    organization = data["data"]["installtion"]["organization"]["slug"]
    app = data["data"]["installtion"]["app"]["slug"]

    send_message(f"{username} 卸载了在 {organization} 的 {app}")


def event_alert(data: dict) -> None:
    pass


def issue(data: dict) -> None:
    username = data["actor"]["name"]
    action = {"created": "创建", "resolved": "解决", "assigned": "分配", "ignored": "忽略"}[
        data["action"]
    ]
    name = data["data"]["issue"]["title"]
    project = data["data"]["issue"]["project"]["name"]
    url = (
        SENTRY_URL + "/organizations/sentry/issues/" + data["data"]["issue"]["id"] + "/"
    )
    send_message(f"{username}{action}了{project}中的issue：[{name}]({url})")


def error(data: dict) -> None:
    username = data["actor"]["name"]
    action = {"created": "创建", "resolved": "解决", "assigned": "分配", "ignored": "忽略"}[
        data["action"]
    ]
    name = data["data"]["error"]["title"]
    url = data["data"]["error"]["web_url"]
    send_message(f"{username}{action}了 error：[{name}]({url})")


functions = {
    "installation": installation,
    "uninstallation": uninstallation,
    "event_alert": event_alert,
    "issue": issue,
    "error": error,
}


@app.post("/")
def handle_sentry():
    signature = hmac.new(
        key=SENTRY_TOKEN.encode("utf-8"), msg=request.body.read(), digestmod=sha256,
    ).hexdigest()
    if not hmac.compare_digest(
        str(signature), str(request.headers["Sentry-Hook-Signature"])
    ):
        raise HTTPError(status=401)

    functions[request.headers["Sentry-Hook-Resource"]](request.json)
    return "everything be ok"

环境变量

注意这里有三个环境变量需要在部署时配置：

SENTRY_TOKEN：Sentry WebHook 配置里的 Client Secret
WECHAT_BOT_URL：企业微信机器人的 POST 接口
SENTRY_URL：自建 Sentry 服务的地址，譬如 https://sentry.example.com 这种。

函数配置

把 HTTP 触发器设置为允许 POST 方法，再把 handler 指定到 index.app 就完事了。

Sentry 配置

最后把 HTTP 触发器对应的地址粘贴到 Sentry 的 WebHook 配置地址里。