Add example script triggering an alert group/escalation per user (#4809)

Related to
https://raintank-corp.slack.com/archives/C0229FD3CE9/p1723474207141749

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/
 

tools/scripts/page_each_user.py

@@ -0,0 +1,127 @@
+# requires requests (pip install requests)
+
+# You can run it like this:
+#    $ ONCALL_API_TOKEN=<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

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