diff --git a/tools/scripts/oncall_reports.py b/tools/scripts/oncall_reports.py index 0084bff844..cccf0fe323 100644 --- a/tools/scripts/oncall_reports.py +++ b/tools/scripts/oncall_reports.py @@ -87,7 +87,7 @@ def _serialize_step(p): teams[t["id"]] = t["name"] -# fetch users (TODO: handle pagination) +# fetch users # https://grafana.com/docs/grafana-cloud/alerting-and-irm/oncall/oncall-api-reference/users/#list-users # GET {{API_URL}}/api/v1/users/ diff --git a/tools/scripts/page_each_user.py b/tools/scripts/page_each_user.py new file mode 100644 index 0000000000..f0eea6998b --- /dev/null +++ b/tools/scripts/page_each_user.py @@ -0,0 +1,127 @@ +# requires requests (pip install requests) + +# You can run it like this: +# $ ONCALL_API_TOKEN= python page_each_user.py + +# This script will create an escalation chain, an escalation policy and a webhook integration +# to trigger alerts to each user in the organization. It will iterate over all users and update +# the escalation policy to notify each user, then trigger an alert group to page that user. + +# By default the escalation chain will be named "Page each user", the integration will be named "Page each user". +# You can customize these names by setting the environment variables ESCALATION_NAME and INTEGRATION_NAME. +# NOTE: You need to remove the existing escalation chain and integration if you want to run this script again. + + +import os +import time + +import requests + +ONCALL_API_BASE_URL = os.environ.get( + "ONCALL_API_BASE_URL", + "https://oncall-prod-us-central-0.grafana.net/oncall", +) +ONCALL_API_TOKEN = os.environ.get("ONCALL_API_TOKEN") +ESCALATION_NAME = os.environ.get("ESCALATION_NAME", "Page each user") +INTEGRATION_NAME = os.environ.get("INTEGRATION_NAME", "Page each user") + +headers = { + "Authorization": ONCALL_API_TOKEN, +} + + +def setup_escalation(): + """Setup an escalation chain to be used by the paging integration.""" + response = requests.post( + f"{ONCALL_API_BASE_URL}/api/v1/escalation_chains", + headers=headers, + json={"name": ESCALATION_NAME}, + ) + response.raise_for_status() + return response.json() + + +def setup_escalation_policy(escalation_chain): + """Setup a base escalation policy associated to the given escalation chain.""" + response = requests.post( + f"{ONCALL_API_BASE_URL}/api/v1/escalation_policies", + headers=headers, + json={ + "escalation_chain_id": escalation_chain["id"], + "type": "wait", + "duration": 60, + }, + ) + response.raise_for_status() + return response.json() + + +def update_escalation_to_notify_user(escalation_policy, user): + """Update the escalation policy to notify the given user.""" + response = requests.put( + f"{ONCALL_API_BASE_URL}/api/v1/escalation_policies/{escalation_policy['id']}", + headers=headers, + json={ + "type": "notify_persons", + "persons_to_notify": [user["id"]], + }, + ) + response.raise_for_status() + + +def setup_integration(escalation_chain): + """Setup a webhook integration to trigger alerts following the given escalation chain.""" + response = requests.post( + f"{ONCALL_API_BASE_URL}/api/v1/integrations", + headers=headers, + json={ + "name": INTEGRATION_NAME, + "type": "webhook", + "default_route": { + "escalation_chain_id": escalation_chain["id"], + }, + "templates": { + "web": { + "title": "{{ payload.title }}", + } + } + }, + ) + response.raise_for_status() + return response.json() + + +# setup escalation chain, escalation policy and integration + +escalation_chain = setup_escalation() +escalation_policy = setup_escalation_policy(escalation_chain) +integration = setup_integration(escalation_chain) + +# iterate users, update escalation policy and trigger alert group +page = 1 +while True: + url = ONCALL_API_BASE_URL + "/api/v1/users/" + r = requests.get(url, params={"page": page}, headers=headers) + r.raise_for_status() + response_data = r.json() + results = response_data.get("results") + for u in results: + print("Updating escalation for user", u["username"]) + update_escalation_to_notify_user(escalation_policy, u) + + print("Triggering alert group for user", u["username"]) + response = requests.post( + integration["link"], + headers=headers, + json={ + "title": f"Paging user {u['username']}", + "message": "Please acknowledge this alert" + }, + ) + # wait a bit to avoid rate limiting (and allow alert processing before next one) + time.sleep(5) + + page += 1 + total_pages = int(response_data.get("total_pages")) + if page > total_pages: + break diff --git a/tools/scripts/readme.md b/tools/scripts/readme.md index a269609791..a0503844db 100644 --- a/tools/scripts/readme.md +++ b/tools/scripts/readme.md @@ -1,19 +1,22 @@ # Sample scripts using public API -- [oncall_hours_reports.py](oncall_hours_reports.py) +- [oncall_hours_reports.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/oncall_hours_report.py) Generate per-user on-call hours report -- [oncall_reports.py](oncall_reports.py) +- [oncall_reports.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/oncall_reports.py) Generate CSV user reports using public API -- [shift_shifts.py](shift_shifts.py) +- [page_each_user.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/page_each_user.py) +Create an integration and trigger an alert group per user targeting each user + +- [shift_shifts.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/shift_shifts.py) Shift schedule shifts by a given delta -- [mattermost_webhooks.py](mattermost_webhooks.py) +- [mattermost_webhooks.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/mattermost_webhooks.py) Setup Mattermost webhooks for alert group notifications -- [discord_webhooks.py](discord_webhooks.py) +- [discord_webhooks.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/discord_webhooks.py) Setup Discord webhooks for alert group notifications -- [swap_requests_workday.py](swap_requests_workday.py) +- [swap_requests_workday.py](https://github.com/grafana/oncall/blob/dev/tools/scripts/swap_requests_workday.py) Create shift swap requests using Workday absences information