はじめに

現職ではAWSをメインに扱うクラウドインテグレーターでエンジニアをしております。
お客様毎によって異なるAWSアカウントを構築、運用しております。
オプションとして監視契約を従量課金の料金体系にする試みがありました。

現職の監視ではCloudWatchのアラートをPager Dutyに連携して通知する仕組みを利用しています。
そこでPager Duty APIを利用して対象AWSアカウントの月間アラート件数を毎月自動で取得する設定を行いました。
コーディングには勉強中のPythonを使いました。
Pager DutyのAPIの日本語記事が少なかった事、自分の忘備録として記載致します。
改善点や何かありましたらコメントを頂けると嬉しく思います。

Pager Dutyとは

PagerDutyは様々な監視ツールと連携して、システムのインシデントを一元管理するツールです。
現職での主な利用用途はCloudWatch AlarmのSNSから通知を受け取りエンジニアのPager Dutyアプリに通知させる目的です。
Pager Dutyアプリでは電話のように音がなるのでメール通知よりも気づけるようになります。

Pager Duty API　

まずはAPIを利用するためのトークンを取得します。
Pager Dutyにログインしてから My Profileに進みます。
User Settingの下部にAPI Accessの項目があります。
Create API User Tokenのボタンがありますのでこちらからキーを生成します。

APIドキュメントを参考にアラート履歴を確認するために必要な流れを確認します。
Pager Dutyの通知内容を確認してみるとAlerts>CUSTOM DETAILS>AccountにAWSアカウントIDが記載されていました。
そのため月間のAlertsを全件取得してAlerts IDリスト作成、Alerts IDリストをfor文で繰り返して特定Accountの値をカウントする事で実現可能です。

どうやらPager Dutyでは監視サーバーから受け取った通知をINCIDENTSという単位で管理しており、各INCIDENTSの一項目としてAlertsという内容が含まれているようです。

改めてまとめると以下の流れでカウントします。

インシデントのリストを作成
インシデントのリストからAlertsのリストを作成
Alertsのリストから特定アカウントのIDが含まれている場合はカウントする
カウントした数を通知する

ソースコード

ここではcountした数をprintして確認する内容で記載しております。
実際にはSlackへcountした数を投稿するようにCloudWatch EventsでLambdaを定期実行しています。

import requests
import json
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta

#抽出期間の前月と今月を自動で取得する
today = datetime.today()
one_month_before_tmp = today + relativedelta(months=-1)
one_month_before = datetime.strftime(one_month_before_tmp, '%Y-%m')
this_month =datetime.strftime(today , '%Y-%m')

#１ヶ月間のインシデントのリストを作成
account_id = '確認したいAWSアカウントID'
api_key = 'APIキー'
incidents = []
offset=0

while True:
    url = 'https://api.pagerduty.com/incidents'
    headers = {'Accept': 'application/vnd.pagerduty+json;version=2', 'Authorization': f'Token token={api_key}','Content-Type':'application/json'}
    payload = {'time_zone': 'Asia/Tokyo','since': f'{one_month_before}-25','until':f'{this_month}-25','limit':100,'offset':offset}
    incidents_response = requests.get(url, headers=headers,params=payload)
    incidents_response = incidents_response.json()
    
    for incident in incidents_response["incidents"]:
        incidents.append(incident["id"])

    offset += 100
    hasMore = incidents_response['more']
    if not hasMore:
        break

#インシデントリストを元にアカウント毎のアラート数リストを作成する
#現職の事情からアカウントIDが入っていない通知の場合はpass（リージョン障害等です）
accounts = []
for id in incidents:
    url = f'https://api.pagerduty.com/incidents/{id}/alerts'
    alerts_response = requests.get(url, headers=headers)
    alerts_response = alerts_response.json()
    try:
        account = alerts_response['alerts'][0]['body']['cef_details']['details']['Account']
        accounts.append(account)
    except:
        pass

#対象AWSアカウントIDが何回発生したかを確認する
count = (accounts.count(account_id))
print(count)

ページング

APIでは一度で取得できる内容に制限があったりします。
Pager Dutyのインシデント数確認では上限が100件でした。
101件以上の内容を全て取得するためにページングという概念があります。
複数のページにまたがった本をペラペラめくるイメージを持ちました。
一枚のページには100の項目までで、101〜は2ページを見るといったところでしょうか。

クエリでoffsetオプションを利用することで特定のページを開く事ができます。
デフォルト（0）だと最初の１つ目から取得しますがoffset:100と指定すると101件目から200まで取得可能です。
続きのページがある時はresponseの['more']がTrueとなって知らせてくれます。

そのため、responseの['more']がTrueの間、offsetに100を足してPager Dutyのインシデント数を確認する、という処理にすることで特定期間のアラートを複数のページにまたがって取得することが可能です。

おわりに

Pager Dutyは便利な割に日本語のドキュメントが少なくプログラム初学者の自分には苦労するところがありました。
この記事がどなたかの参考になれば幸いです。