さまざまな機能や統合が充実しているPagerDuty。APIを使うと更に痒いところに手が届くので
自分の経験からいくらかピックアップしてまとめてみました。何かの参考になれば幸いです！

① PagerDuty API概要
② API利用時に気にかけていること
③ 活用例

という流れで記載します。

①PagerDuty API概要

はじめかた

APIリファレンス https://developer.pagerduty.com/api-reference
APIキーの発行 https://support.pagerduty.com/lang-ja/docs/api-access-keys
呼び出し方
- リファレンスにcurlでの呼び出しサンプルあり
- SDK: https://developer.pagerduty.com/docs/ZG9jOjExMDI5NTg2-api-client-libraries
- Tools & Code Samples: https://developer.pagerduty.com/docs/ZG9jOjExMDI5NTg3-api-tools-and-code-samples

(例) インシデントを取得してみる

リスト取得

リファレンス: https://developer.pagerduty.com/api-reference/9d0b4b12e36f9-list-incidents

curl --request GET \
  --url https://api.pagerduty.com/incidents \
  --header 'Accept: application/json' \
  --header 'Authorization: Token token=★発行したAPIキーを入れる★' \
  --header 'Content-Type: application/json'

一度の取得でとれる限界数(limit)を超えて取得するときはoffsetを使ってページング呼び出しします。
SDKの中にはページング処理をラップしてくれているものもあります。

指定インシデントの取得

リファレンス: https://developer.pagerduty.com/api-reference/005299ed43553-get-an-incident

curl --request GET \
  --url https://api.pagerduty.com/incidents/★取得したいインシデントIDを入れる★ \
  --header 'Accept: application/json' \
  --header 'Authorization: Token token=★発行したAPIキーを入れる★' \
  --header 'Content-Type: application/json'

②API利用時に気にかけていること

テスト

APIを叩くスクリプトを作成して動作確認するとき、影響範囲を考慮し以下のようにしています。

本番とは別に検証用PagerDutyがあれば、そちらで先に試す
ループで処理を繰り返す場合も、まずはループ数を制限したりSleepを入れる形で徐々に実行
- テストで意図せずAPI呼び出しレート(後述)を消費しつくさないための考慮も兼ねる
Writeするような処理は慎重にテストする
Writeするような処理は変更前の設定値をバックアップとっておく

API呼び出しレート

https://developer.pagerduty.com/docs/72d3b724589e3-rest-api-rate-limits
過度な呼び出しを行うとレートにひっかかるので注意が必要。
レスポンスに含まれるRate limit headersを確認すると呼び出しレートの消費状況やリセットまでの時間が確認できます。

参考： REST API Rate Limits::ベストプラクティス

API権限

Read処理だけの場合はReadOnlyのAPIキーを使う
用途が異なる場合はAPIキーを分ける

③ 活用例

設定の一括変更

特定の文字列を含むサービスに紐づける「エスカレーションポリシー」を一括で同じものにに変更したい！ということがあり、対象サービスが300以上あったのでスクリプトを作成して実施したことがあります。

利用した主なAPI

当時書いたコード

SDK使わず書いたのでリクエストや自前ページングなどかなり長くなっていました。SDK使うとだいぶすっきりしそう。

import sys
import requests
import time
import json

ENDPOINT = "https://api.pagerduty.com/"
LIMIT = 100
SLEEP = 0.1
MAX_OFFSET = 100000

TARGET_SERVICE_WORD = "" # ★特定の文字列をここに入れる
SET_ESCALATION_POLICY = "" # ★あらたに紐づけるエスカレーションポリシーのID


def main(key):
    # 対象サービス取得
    services = get_services(key)

    # 対象サービス表示
    for s in services:
        print(f"{s['id']}\t{s['summary']}\t{s['html_url']}")
    target_count = len(services)
    print(f"{len(services)} services")

    # 確認
    print("ok? y")
    if "y" != input():
        print("canceled")
        exit()

    # バックアップ
    backup_fname = f'backup_{int(time.time())}.json'
    with open(backup_fname, 'w') as f:
        json.dump(services, f, indent=1)
    print(f"saved {backup_fname}")

    # 更新
    count = 0
    for s in services:
        count += 1
        print(f"{count}/{target_count} update {s['id']}\t{s['summary']}")
        change_escalation_policy(s, key, SET_ESCALATION_POLICY)
        time.sleep(SLEEP)

    print("finished")


def change_escalation_policy(service, api_key, policy_id):
    res = requests.put(
        url=f"{ENDPOINT}/services/{service['id']}",
        headers={
            "Content-Type": "application/json",
            "Authorization": "Token token=" + api_key,
            "Accept": "application/vnd.pagerduty+json;version=2",
            "From": "" #★書き込みユーザーアカウント
        },
        data=json.dumps({
            "service": {
                "type": "service",
                "escalation_policy": {
                    "id": policy_id,
                    "type": "escalation_policy_reference"
                }
            }
        })
    ).json()
    return res["service"]


def get_services(api_key):
    offset = 0
    services = []
    more = True
    while more:
        if offset > 0:
            time.sleep(SLEEP)

        # request
        print(f"request offset:{offset} limit:{LIMIT}")
        res = requests.get(
            url=f"{ENDPOINT}/services",
            headers={
                "Content-Type": "application/json",
                "Authorization": "Token token=" + api_key,
                "Accept": "application/vnd.pagerduty+json;version=2"
            },
            params={
                "offset": offset,
                "limit": LIMIT
            }
        ).json()

        # 次の実行準備
        more = res.get("more")
        count = len(res["services"])
        offset += count
        print(f"  count={count}, more={more}")

        # 条件に一致するサービスを抽出
        target_services = [
            s
            for s in res["services"]
            if is_target(s)
        ]
        print(f"found target services={len(target_services)}")
        services.extend(target_services)
        if offset > MAX_OFFSET:
            print("over MAX_OFFSET")
            exit()

    return services


def is_target(service):
    summary = service.get("summary", None)
    if summary is None:
        return False
    return TARGET_SERVICE_WORD in summary


if __name__ == '__main__':
    main(key=sys.argv[1])

③活用例

インシデントの対応自動化

起票されたインシデントの情報をWebhookで受け、適切な一次対応(情報取得、復旧作業、エスカレーション）を自動で行うインハウスシステムを開発して、かれこれ6年ちかく運用しています。

「インシデントのステータス更新」「一次対応結果をノートやカスタムフィールドに書き込み」などの場面でAPIを利用しています。

利用した主なAPI

詳細

「PagerDuty」を活用した次世代監視システムの開発

分析基盤へのデータのロード

PagerDutyは分析機能が充実していますが、社内の分析基盤とも統合し運用の改善や評価に役立てています。
インシデントデータを定期的に取り込んでデータソースに蓄積しています。

利用した主なAPI

List incidents

詳細

インシデント対応品質を高める「運用分析プラットフォーム」を短期間で構築

終わりに

他にもちょっとした作業や分析でスクリプトを書いて対応することはあります。

PagerDutyはアップデートが頻繁にあるので「これ自前でスクリプト書いてたけど、この追加された仕組みを使った方がいいな」となってそちらに切り替えたこともあるので、アップデートも追いかけていきたいですね。

ではまた！