Sentry: Automate setup of alerting rules for a project

sentry-alert.py

import argparse
import json
import os
import requests

ORG_SLUG = "..."
SLACK_WORKSPACE_ID = "..."
OWNER_TEAM = "..."

SLACK_RULES = {
    "slack-ops": {
        "name": "slack-ops",
        "filters": [{"id": "sentry.rules.filters.level.LevelFilter", "match": "gte", "level": "40"}],
        "channel_name": "ops",
        "environment": "production",
    },
    "slack-ops-other": {
        "name": "slack-ops-other",
        "filters": [
            {"id": "sentry.rules.filters.level.LevelFilter", "match": "gte", "level": "30"},
            {
                "attribute": "environment",
                "match": "ne",
                "value": "production",
                "id": "sentry.rules.filters.event_attribute.EventAttributeFilter",
            },
        ],
        "channel_name": "ops-other",
        "environment": None,
    },
}


def rule_payload(rule: dict, workspace_id: str = SLACK_WORKSPACE_ID, frequency=720, owner=OWNER_TEAM):
    return {
        "conditions": [{"id": "sentry.rules.conditions.every_event.EveryEventCondition"}],
        "filters": rule["filters"],
        "actions": [
            {
                "tags": "environment,url",
                "workspace": workspace_id,
                "id": "sentry.integrations.slack.notify_action.SlackNotifyServiceAction",
                "channel": rule["channel_name"],
            }
        ],
        "actionMatch": "all",
        "filterMatch": "all",
        "frequency": frequency,
        "name": rule["name"],
        "owner": owner,
        "environment": rule["environment"],
    }


def _request(path, method="post", **kwargs):
    AUTH_TOKEN = os.getenv("SENTRY_TOKEN")  # A token with scopes project:read,project:write fetched from
    assert AUTH_TOKEN
    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}

    url = f"https://sentry.io/api/0{path}"
    res = requests.request(method, url, headers=headers, **kwargs)
    res.raise_for_status()
    return res


def list_projects():
    path = "/projects/"
    res = _request(path, method="get")
    return res.json()


def list_rules(app: str, org: str):
    path = f"/projects/{org}/{app}/rules/"
    res = _request(path, method="get")
    return res.json()


def fetch_rule(app: str, rule_id: int, org: str):
    path = f"/projects/{org}/{app}/rules/{rule_id}/"
    res = _request(path, method="get")
    return res.json()


def create_rule(app: str, data: dict, org: str):
    path = f"/projects/{org}/{app}/rules/"
    res = _request(path, json=data)
    return res.json()


def delete_rule(app: str, rule_id: int, org: str):
    path = f"/projects/{org}/{app}/rules/{rule_id}/"
    try:
        _request(path, method="delete")
    except requests.exceptions.HTTPError:
        return False
    return True


def main(apps: list[str], all_apps: bool, create_rules: bool, prune_rules: bool, list_apps: bool, org: str):
    if list_apps:
        print(json.dumps([project["slug"] for project in list_projects()], indent=2))
        return

    if all_apps:
        apps = [project["slug"] for project in list_projects()]

    for app in apps:
        fresh_rules = list_rules(app, org)
        existing = {rule["name"] for rule in fresh_rules}

        if create_rules:
            for name, rule in SLACK_RULES.items():
                if name in existing:
                    print(f"Rule with {name=} already exists, skipping...")
                    continue

                data = rule_payload(rule)
                create_rule(app, data, org)

        if prune_rules:
            to_delete = existing - set(SLACK_RULES.keys())
            for rule in to_delete:
                rule_id = [r for r in fresh_rules if r["name"] == rule][0]["id"]
                delete_rule(app, rule_id, org)

        if not create_rules and not prune_rules:
            print(json.dumps(fresh_rules, indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Update sentry project with a set of standard rules")
    parser.add_argument("--apps", nargs="*")
    parser.add_argument("--list-apps", action="store_true")
    parser.add_argument("--all", action="store_true")
    parser.add_argument("--create", action="store_true")
    parser.add_argument("--prune", action="store_true")
    parser.add_argument("--org", default=ORG_SLUG)
    args = parser.parse_args()
    main(args.apps, args.all, args.create, args.prune, args.list_apps, args.org)