はじめに

本記事では、Amazon ElastiCache のイベント通知を Amazon SNS で受け取り、Lambda 経由で PagerDuty にインシデントとして起票する構成を、実際の設定手順とあわせて紹介します。

ElastiCache のイベントについて

ElastiCacheのイベントを通知するには、Amazon SNSでの通知と、EventBridgeでの通知が利用できます。
ただ、EventBridge に届く ElastiCache イベントの一覧は以下のみで、ElastiCache のフェイルオーバーイベントは含まれていません。そのため、今回は Amazon SNS で通知する構成を組んでいきます。

※ EventBridge に届く ElastiCache イベントの一覧

  • Cache Created / Cache Creation Failed / Cache Deleted
  • Cache Updated / Cache Update Failed
  • Cache Limit Approaching
  • Snapshot Created / Snapshot Creation Failed
  • Snapshot Export Failed / Snapshot Copy Failed

全体構成

通知が届くまでの流れは次のとおりです。

[ElastiCache]
     │
     ▼
[SNS トピック]
     │
     └─（サブスクリプションフィルター）→ [Lambda] → [PagerDuty Events API v2]

今回、 Amazon SNS のサブスクリプションフィルターで必要なイベントだけを絞り込んでいます。ElastiCache はクラスター作成・削除・スケーリング・メンテナンスなど、あらゆるイベントを同じトピックに発行してくるため、フィルターなしでは PagerDuty に多数のインシデントが通知される可能性があります。

また、Amazon SNS から PagerDuty に直接通知することも可能ですが、通知タイトルが「AWS Notification Message」となり分かりづらいので、Lambdaを使用してタイトルを編集しています。( PagerDuty の Event Orchestrationでもフィルタリングやインシデントタイトルの編集ができますが、今回は使用しません。）

前提条件

  • PagerDuty 側でサービスを作成し、Events API v2 のインテグレーションキー（routing_key）を払い出し済み
  • SNS トピック / Lambda 関数 / IAM ロール / Secrets Manager シークレットを作成できる IAM 権限を保有している

検知対象のイベント

イベント名 概要
ElastiCache:FailoverComplete レプリカノードへのフェイルオーバーが完了（Valkey / Redis OSS のみ）

手順

① SNS トピックの作成

  1. [Amazon SNS] → [トピック] → [トピックの作成] を押下
  2. 以下を設定してトピックを作成します
項目 値 備考
タイプ スタンダード —
名前 例) elasticache-event —
表示名 例) ElastiCache Event —
暗号化 無効 有効にすると ElastiCache から発行できません

トピックを作る際の制約がいくつかあります。いずれも AWS ドキュメントに明記されているものです。

  • 暗号化されたトピックには発行できない — KMS で暗号化を有効にすると、ElastiCache 側からの発行が失敗します
  • 1 つのクラスターに設定できるトピックは 1 つだけ — 通知先を分けたい場合は、トピックの先でサブスクリプションを分岐させます
  • トピックはクラスターと同一リージョン・同一アカウントである必要がある

3. 作成したトピックの [アクセスポリシー] を編集し、ElastiCache からの発行を許可します

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowElastiCachePublish",
      "Effect": "Allow",
      "Principal": {
        "Service": "elasticache.amazonaws.com"
      },
      "Action": "SNS:Publish",
      "Resource": "arn:aws:sns:＜リージョン＞:＜AWSアカウントID＞:＜トピック名＞",
      "Condition": {
        "StringEquals": {
          "aws:SourceOwner": "＜AWSアカウントID＞"
        }
      }
    }
  ]
}

既定のアクセスポリシー（アカウント内からの各種操作を許可するステートメント）は残したまま、上記ステートメントを追加してください。

② ElastiCache 側で SNS 通知を有効化

  1. [ElastiCache] → 対象のクラスター（レプリケーショングループ）を選択 → [変更] を押下
  2. [Amazon SNS 通知] の [トピック] で、①で作成した SNS トピックを選択
  3. [変更をプレビュー] → [変更] で反映

③ PagerDuty の認証情報を Secrets Manager に登録

PagerDuty のインテグレーションキーは、Lambda の環境変数に直書きせず Secrets Manager で管理します。

  1. [AWS Secrets Manager] → [新しいシークレットを保存する] を押下
  2. シークレットのタイプで [その他のシークレットのタイプ] を選択し、以下のキー / 値を登録
キー 値
pagerduty_routing_key PagerDuty サービスのインテグレーションキー

3. シークレット名を設定して保存します（例: elasticache-event-notification）

④ Lambda 実行ロールの作成

  1. [IAM] → [ロール] → [ロールを作成] を押下
  2. 信頼されたエンティティ: [AWS のサービス] → [Lambda]
  3. 以下の権限を付与します
ポリシー 用途
AWSLambdaBasicExecutionRole（AWS 管理ポリシー） CloudWatch Logs への出力
インラインポリシー（下記） Secrets Manager からの認証情報取得 
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:＜リージョン＞:＜AWSアカウントID＞:secret:＜シークレット名＞-*"
    }
  ]
}

⑤ Lambda 関数の作成

  1. [AWS Lambda] → [関数の作成] → [一から作成] を押下
  2. 以下を設定します
項目 値 備考
関数名 例) elasticache-event-notification —
ランタイム Python 3.13 —
アーキテクチャ x86_64 —
実行ロール ④で作成したロールを選択 —
タイムアウト 1 分

3. [設定] → [環境変数] に以下を設定します

キー 値
NOTIFICATION_SECRET_ID ③で作成したシークレットの名前または ARN
PD_SEVERITY critical（省略時は critical）

4. 以下のコードをデプロイします

import json
import os
import urllib.error
import urllib.request

import boto3

TITLE_TEMPLATE = "[elasticache-redis-failover発生] 対象:{target}"
EVENT_KEY_PREFIX = "ElastiCache:"
SECRET_ID = os.environ["NOTIFICATION_SECRET_ID"]
PD_SEVERITY = os.environ.get("PD_SEVERITY", "critical")
PD_ENDPOINT = "https://events.pagerduty.com/v2/enqueue"

_secret_cache = None

def _get_secret():
    global _secret_cache
    if _secret_cache is None:
        client = boto3.client("secretsmanager")
        response = client.get_secret_value(SecretId=SECRET_ID)
        _secret_cache = json.loads(response["SecretString"])
    return _secret_cache


def _parse_message(raw_message):
    try:
        body = json.loads(raw_message)
    except json.JSONDecodeError:
        body = None

    if not isinstance(body, dict) or not body:
        return "ElastiCache:Event", "", {"Message": raw_message}

    def _as_text(value):
        return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)

    for key, value in body.items():
        if key.startswith(EVENT_KEY_PREFIX):
            return key, _as_text(value), body

    event_name, value = next(iter(body.items()))
    return event_name, _as_text(value), body


def _extract_target(event_value):
    value = event_value.strip()
    target = value.rsplit(":", 1)[-1].strip() if ":" in value else value
    return target or "unknown"


def _region_from_arn(arn):
    parts = arn.split(":")
    return parts[3] if len(parts) > 3 else ""


def _build_title(target):
    return TITLE_TEMPLATE.format(target=target)


def _post_json(url, payload, headers=None):
    data = json.dumps(payload).encode("utf-8")
    request_headers = {"Content-Type": "application/json"}
    if headers:
        request_headers.update(headers)
    req = urllib.request.Request(url, data=data, headers=request_headers, method="POST")
    with urllib.request.urlopen(req, timeout=10) as response:
        return response.status, response.read().decode("utf-8")


def _notify_pagerduty(routing_key, title, target, event_name, body, sns):
    payload = {
        "routing_key": routing_key,
        "event_action": "trigger",
        "dedup_key": sns["MessageId"],
        "payload": {
            "summary": title,
            "source": sns["TopicArn"],
            "severity": PD_SEVERITY,
            "timestamp": sns["Timestamp"],
            "custom_details": {
                "Service": "ElastiCache",
                "EventName": event_name,
                "Target": target,
                "Region": _region_from_arn(sns["TopicArn"]),
                "EventMessage": body,
                "TopicArn": sns["TopicArn"],
                "SnsMessageId": sns["MessageId"],
            },
        },
    }
    return _post_json(PD_ENDPOINT, payload)


def lambda_handler(event, context):
    secret = _get_secret()
    errors = []

    for record in event["Records"]:
        sns = record["Sns"]
        event_name, event_value, body = _parse_message(sns["Message"])
        target = _extract_target(event_value)
        title = _build_title(target)
        print(json.dumps({"event_name": event_name, "title": title}, ensure_ascii=False))

        routing_key = secret.get("pagerduty_routing_key") or ""
        if not routing_key:
            print("pagerduty: skipped (credential is not set)")
            continue

        try:
            status, response_body = _notify_pagerduty(
                routing_key, title, target, event_name, body, sns
            )
            print(f"pagerduty: status={status} body={response_body}")
        except urllib.error.HTTPError as exc:
            errors.append(f"pagerduty: HTTP {exc.code} {exc.read().decode('utf-8')}")
        except Exception as exc:  # noqa: BLE001
            errors.append(f"pagerduty: {exc!r}")

    if errors:
        raise RuntimeError("; ".join(errors))

    return {"status": "ok"}

⑥ サブスクリプションとフィルターポリシーの設定

(1) Lambda サブスクリプションの作成

  1. [Amazon SNS] → [トピック] → ①のトピック → [サブスクリプションの作成] を押下
  2. 以下を設定します
項目 値 備考
プロトコル AWS Lambda —
エンドポイント ⑤で作成した Lambda 関数を選択 —
サブスクリプションフィルターポリシー 有効 下記 (2) 参照
フィルターポリシーのスコープ メッセージ本文 既定の「メッセージ属性」では動作しません

(2) サブスクリプションフィルターポリシー

前述のとおり、ElastiCache はフェイルオーバー以外のイベント（作成 / 削除 / スケーリング / メンテナンス等）も同一トピックに発行します。フェイルオーバーイベントのみを通知対象とするため、以下のフィルターポリシーを設定します。

{
  "ElastiCache:FailoverComplete": [
    {
      "exists": true
    }
  ]
}

ElastiCache のメッセージ本文はイベント名がそのまま JSON のキーになるため、exists 演算子でキーの有無を判定するだけで絞り込めます。

なお、Lambda 側でイベント名を判定して弾く方法もありますが、今回は SNS のフィルターポリシーで絞ります。

⑦ 動作確認

ElastiCache に対してフェイルオーバーを発生させたところ、PagerDuty に想定どおりの内容でインシデントが起票されました。

まとめ

今回、ElastiCache のイベント通知を Amazon SNS で受け取り、Lambda 経由で PagerDuty にインシデントとして起票する手順を紹介しました。
最後までご覧いただきありがとうございました。