まえがき

この記事はPagerDuty Advent Calendarの14日目になります。


AWS LambdaとPagerDuty APIを組み合わせて、
6ヶ月以上前のインシデントデータを取得する方法について解説します。

PagerDutyは、多くの監視ツールとのインテグレーションに対応しており、様々なインシデントの一括管理が可能なSaaSです。
PagerDutyには、日々のインシデントデータが集約されています。しかし、APIを介してこれらの過去のデータにアクセスしようとすると、特定の制限に直面することがあることが分かりました。
今回この制限に対応し過去のインシデントデータにアクセスするためのアプローチをご紹介したいと思います。

目次

①PagerDuty APIの解説

PagerDuty APIでは、インシデントの作成・更新・などの基本的な操作が可能です。
特に今回は過去のインシデントデータを取得するため、APIドキュメントのList incidentsを参照しました。

PagerDutyのAPIドキュメントの参照画像

上述の画像より、過去のインシデントデータを取得するためのリクエスト送信には、以下情報が必要とわかりました。

ヘッダー情報
Authorization: APIキー
↳後述の手順内でAPIキーを作成します。
Accept: application/vnd.pagerduty+json;version=2
Content-Type: application/json
required(必須)と記載があります。

▼インシデントデータ取得の制限について

さらにAPI経由で過去のインシデントデータを参照する制限として、sinceとuntilパラメーターで指定すれば最大6ヶ月間、指定しない場合デフォルトで1ヶ月間が適用されるということが分かりました。

since
↳ 検索したい日付範囲の開始日。最大範囲は6ヶ月で、デフォルトは1ヶ月です。
until
↳ 検索したい日付範囲の終了日。最大範囲は6ヶ月、デフォルトは1ヶ月です。

つまりAPIリスクエスト送信時にsinceとuntilパラーメーターを使用することで、過去のインシデントデータを参照することが可能となります。
しかしsinceとuntilパラーメーターを使用しない場合は、過去1ヶ月間しか参照ができない仕様にも見受けられます。

以降のセクションでは、実際に環境構築しsinceとuntilパラーメーターの有無でどんなインシデントデータが取得可能なのか?検証をしていきます。

②環境構築

・Lambdaを作成します。

項目
Lambda名 pd-list-function(なんでもOK)
Runtime Python3.11
Architecture x86_64
Layer 「※1の手順:」で作成して追加

※1の手順:
外部のPagerDuty APIにアクセスする必要があるためrequestsライブラリを使えるように、【1】ローカル環境で、zipファイルを作成する。を参照してレイヤーをLambda関数に適用します。

③PagerDuty APIを叩いて最新インシデントデータを取得(since~until未使用)

③-1 PagerDutyに存在する過去のインシデントデータを確認

サンプルのPagerDutyインシデントデータには、2023/12/1に2件。そして2022/4/11に1件ずつ存在していました。
今回は、2023/12/1に2件を取得します。上述で確認したPagerDutyのAPIドキュメント通りだと2022/4/11のデータは取得できないはずです。
よって期待の出力結果は以下になります。

期待値:
・「2023/12/1のインシデントデータ」のみ取得できる。
・「2022/4/11のインシデントデータ」は取得できない。

③-2 Lambda環境変数を更新する。

項目
PAGERDUTY_API_KEY 「※2の手順:」で作成して追加
PAGERDUTY_ENDPOINT https://api.pagerduty.com/incidents

※2の手順:
・PagerDutyにログインし、上部タブ「Integrations」→「API Access Keys」をクリック。(以下の画像を参照)
・「Create New API Key」→「好きなAPIキー名」入力→「Create Key」をクリック。(以下の画像を参照)
・作成された画面にAPIキーの値が表示されるのでコピペして、Lambda環境変数へ貼り付け。


③-3 Lambda関数のコードにコピペして「test」から実行する。

※コードの詳細な解説は省略します。

import os
import json
import requests
import logging

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 環境変数の設定
PAGERDUTY_API_KEY = os.environ["PAGERDUTY_API_KEY"]
PAGERDUTY_ENDPOINT = os.environ["PAGERDUTY_ENDPOINT"]

def lambda_handler(event, context):
    try:
        # PagerDutyから特定サービスIDの最近のインシデントを取得
        incidents = fetch_recent_incidents()
        # インシデント情報を指定された形式で整形
        formatted_incidents = format_incidents(incidents)
        formatted_output = "\n".join(formatted_incidents)
        # 結果をログに出力
        logger.info(formatted_output)
        return {
            "statusCode": 200,
            "body": formatted_output,
            "headers": {
                "Content-Type": "text/plain"
            }
        }
    except Exception as e:
        logger.error(f"Error: {e}")
        return {
            "statusCode": 500, 
            "body": str(e), 
            "headers": {
                "Content-Type": "text/plain"
            }
        }

def fetch_recent_incidents():
    headers = {
        "Authorization": f"Token token={PAGERDUTY_API_KEY}",
        "Accept": "application/vnd.pagerduty+json;version=2",
        "Content-Type": "application/json"
    }
    params = {
        "statuses[]": ["resolved"],
        "limit": 10
    }
    response = requests.get(PAGERDUTY_ENDPOINT, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("incidents", [])

def format_incidents(incidents):
    formatted_data = []
    for index, incident in enumerate(incidents, start=1):
        formatted_data.append(
            f"\nNo:{index}\n"
            f"Title: {incident['title']}\n"
            f"Created At: {incident['created_at']}\n"
            f"Status: {incident['status']}\n"
            f"Service: {incident['service']['summary']}\n"
            f"URL: {incident['html_url']}"
        )
    return formatted_data

実行結果:
期待どおり「2023/12/1のインシデントデータ」のみ取得できました。

No:1
Title: 2023-12-01_Example Incident-1
Created At: 2023-12-01T11:54:04Z
Status: resolved
Service: test
URL: https://~省略

No:2
Title: 2023-12-01_Example Incident-2
Created At: 2023-12-01T11:54:04Z
Status: resolved
Service: test
URL: https://~省略

④PagerDuty APIを叩いて過去インシデントデータを取得(since~until使用)

次にsince~untilで2022/4/11を参照するように何日後何日前計算機で計算。
2023/12/1から2022/4/11は599日前だったので、Lambda環境変数に渡して実行します。
Pythonコードのしていることはスクショのようなイメージです。
sinceとuntilを代入して特定日付まで遡ってデータ取得しています。

期待の出力結果は以下の通りです。

期待値:
・「2023/12/1のインシデントデータ」は取得できない。
・「2022/4/11のインシデントデータ」のみ取得できる。

④-1 Lambda環境変数を更新する。

項目
PAGERDUTY_API_KEY 前述手順で作成済
PAGERDUTY_ENDPOINT https://api.pagerduty.com/incidents
SINCE_DAYS_AGO 600
UNTIL_DAYS_AGO 598

④-2 Lambda関数のコードにコピペして「test」から実行する。

※コードの詳細な解説は省略します。

import os
import json
import requests
import datetime
import logging

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 環境変数の設定
PAGERDUTY_API_KEY = os.environ["PAGERDUTY_API_KEY"]
PAGERDUTY_ENDPOINT = os.environ["PAGERDUTY_ENDPOINT"]
UNTIL_DAYS_AGO = int(os.environ["UNTIL_DAYS_AGO"])
SINCE_DAYS_AGO = int(os.environ["SINCE_DAYS_AGO"])

def lambda_handler(event, context):
    try:
        # PagerDutyから特定期間内のインシデントを取得
        incidents = fetch_incidents_for_period()
        # インシデント情報を指定された形式で整形
        formatted_incidents = format_incidents(incidents)
        formatted_output = "\n".join(formatted_incidents)
        # 結果をログに出力
        logger.info(formatted_output)
        return {
            "statusCode": 200,
            "body": formatted_output,
            "headers": {
                "Content-Type": "text/plain"
            }
        }
    except Exception as e:
        logger.error(f"Error: {e}")
        return {
            "statusCode": 500, 
            "body": str(e), 
            "headers": {
                "Content-Type": "text/plain"
            }
        }

def fetch_incidents_for_period():
    until_date = datetime.datetime.utcnow() - datetime.timedelta(days=UNTIL_DAYS_AGO)
    since_date = until_date - datetime.timedelta(days=SINCE_DAYS_AGO)

    # 最大範囲を6ヶ月(約180日)に制限
    max_range = datetime.timedelta(days=180)
    if since_date < until_date - max_range:
        since_date = until_date - max_range

    headers = {
        "Authorization": f"Token token={PAGERDUTY_API_KEY}",
        "Accept": "application/vnd.pagerduty+json;version=2",
        "Content-Type": "application/json"
    }
    params = {
        "statuses[]": ["resolved"],
        "since": since_date.isoformat(),
        "until": until_date.isoformat(),
        "limit": 10
    }
    response = requests.get(PAGERDUTY_ENDPOINT, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("incidents", [])

def format_incidents(incidents):
    formatted_data = []
    for index, incident in enumerate(incidents, start=1):
        formatted_data.append(
            f"\nNo:{index}\n"
            f"Title: {incident['title']}\n"
            f"Created At: {incident['created_at']}\n"
            f"Status: {incident['status']}\n"
            f"Service: {incident['service']['summary']}\n"
            f"URL: {incident['html_url']}"
        )
    return formatted_data

実行結果:
期待どおり「1年以上前のインシデントデータ」のみ取得できました。

No:1
Title: Example Incident
Created At: 2022-04-11T03:01:57Z
Status: resolved
Service: geeks-kohei
URL: https://~省略

⑤ まとめ

PagerDutyのAPIを利用し、通常は6ヶ月までの制限がある過去のインシデントデータの取得に成功しました。
sinceとuntilのパラメータを適切に設定することで、通常の制限を超えた期間のデータにもアクセスできることが分かりました。
日々のインシデントをPagerDutyで一括管理していると、6ヶ月以上過去のインシデントデータへを参照する場面は意外と多いのではないでしょうか。
今回まとめた内容がお役に立てば嬉しく感じます。

最後まで閲覧いただきありがとうございました。