Skip to content

Instantly share code, notes, and snippets.

@nikolaik
Last active April 17, 2023 18:29
Show Gist options
  • Save nikolaik/85e19b89223686b9ab560822fb63bc01 to your computer and use it in GitHub Desktop.
Save nikolaik/85e19b89223686b9ab560822fb63bc01 to your computer and use it in GitHub Desktop.
Sentry: Automate setup of alerting rules for a project
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", paginate=False, parse_json=True, **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()
if paginate and res.links.get("next", {}).get("results", False):
params = {"cursor": res.links["next"]["cursor"]}
return res.json() + _request(path, method=method, params=params, **kwargs)
return res.json() if parse_json else res
def list_projects():
path = "/projects/"
res = _request(path, method="get", paginate=True)
return res
def list_rules(app: str, org: str):
path = f"/projects/{org}/{app}/rules/"
res = _request(path, method="get", paginate=True)
return res
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
def create_rule(app: str, data: dict, org: str):
path = f"/projects/{org}/{app}/rules/"
res = _request(path, json=data)
return res
def delete_rule(app: str, rule_id: int, org: str):
path = f"/projects/{org}/{app}/rules/{rule_id}/"
try:
_request(path, method="delete", parse_json=False)
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)
@LordSimal
Copy link

This script only works if you have 100 or less projects because it doesn't check the Link HTTP Header for the paginated links.

@nikolaik
Copy link
Author

This script only works if you have 100 or less projects because it doesn't check the Link HTTP Header for the paginated links.

Added some untested Link HTTP header handling

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment