PagerDutyのAPI利用で更に痒いところに手が届く

pagerduty

Last updated at 2023-12-20Posted at 2023-12-20

さまざまな機能や統合が充実している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])