This Python script creates events in PagerDuty using APIv2.
The following script was taken as a basis.
First you need to create a “Routing Key“, aka “Integration Key“, not to be confused with “API Access Key“, which can be used for any API calls, we only need a key from a specific service.
Go to the service settings, in my case it is called “AWS“, and go to the “Integrations” tab
Select – “Add a new integration”
Set the name and select “Integration Type” -> “Use our API directly” -> “Events API v2” and get “Integration Key”
main.py:
import requests
import json
from typing import Any, Dict, List, Optional
routing_key = 'abc123'
api_url = 'https://events.pagerduty.com/v2/enqueue'
def create_event(
summary: str,
severity: str,
source: str = 'aws',
action: str = 'trigger',
dedup_key: Optional[str] = None,
custom_details: Optional[Any] = None,
group: Optional[str] = None,
component: Optional[str] = None,
class_type: Optional[str] = None,
images: Optional[List[Any]] = None,
links: Optional[List[Any]] = None
) -> Dict:
payload = {
"summary": summary,
"severity": severity,
"source": source,
}
if custom_details is not None:
payload["custom_details"] = custom_details
if component:
payload["component"] = component
if group:
payload["group"] = group
if class_type:
payload["class"] = class_type
actions = ('trigger', 'acknowledge', 'resolve')
if action not in actions:
raise ValueError("Event action must be one of: %s" % ', '.join(actions))
data = {
"event_action": action,
"routing_key": routing_key,
"payload": payload,
}
if dedup_key:
data["dedup_key"] = dedup_key
elif action != 'trigger':
raise ValueError(
"The dedup_key property is required for event_action=%s events, and it must \
be a string."
% action
)
if images is not None:
data["images"] = images
if links is not None:
data["links"] = links
response = requests.post(api_url, json = data)
print(response.content)
def main():
create_event(summary = "Instance is down", severity = 'critical', custom_details = "Instance ID: i-12345")
if __name__ == '__main__':
main()
Change the value of the variable “routing_key” to the value “Integration Key“
In this example, an event with the “critical” level will be created with the “Instance is down” title and “Instance ID: i-12345” will be specified in detail.
The script can also be used to create events based on AWS CloudWatch Alarms without using the AWS PagerDuty integration. To do this, create an SNS topic, a Lambda function and specify the SNS topic as the source for the Lambda function. After that, you can create a CloudWatch Alarm and specify the SNS topic as an action when triggered.
main.py (Lambda + SNS):
import requests
import json
from typing import Any, Dict, List, Optional
routing_key = 'abc123'
api_url = 'https://events.pagerduty.com/v2/enqueue'
def get_message_info():
subject = event["Records"][0]["Sns"]["Subject"]
message = event["Records"][0]["Sns"]["Message"]
return subject, message
def create_event(
summary: str,
severity: str,
source: str = 'cloudwatch',
action: str = 'trigger',
dedup_key: Optional[str] = None,
custom_details: Optional[Any] = None,
group: Optional[str] = None,
component: Optional[str] = None,
class_type: Optional[str] = None,
images: Optional[List[Any]] = None,
links: Optional[List[Any]] = None
) -> Dict:
payload = {
"summary": summary,
"severity": severity,
"source": source,
}
if custom_details is not None:
payload["custom_details"] = custom_details
if component:
payload["component"] = component
if group:
payload["group"] = group
if class_type:
payload["class"] = class_type
actions = ('trigger', 'acknowledge', 'resolve')
if action not in actions:
raise ValueError("Event action must be one of: %s" % ', '.join(actions))
data = {
"event_action": action,
"routing_key": routing_key,
"payload": payload,
}
if dedup_key:
data["dedup_key"] = dedup_key
elif action != 'trigger':
raise ValueError(
"The dedup_key property is required for event_action=%s events, and it must \
be a string."
% action
)
if images is not None:
data["images"] = images
if links is not None:
data["links"] = links
response = requests.post(api_url, json = data)
print(response.content)
def main(event, context):
info = get_message_info()
create_event(summary = info[0], severity='critical', custom_details = info[1])
if __name__ == '__main__':
main()
The Python package “requests” will have to be zipped with the Lambda function, since Lambda doesn’t have it. You can use “virtualenv” for this.