はじめに

弊社ではAWS Healthイベントの確認をSlackで行っています。

とある理由から自前でカスタマイズした通知を実装してみたので、今回はその経緯や実装内容について書きたいと思います。

背景

通常の実装方法

HealthイベントをSlackに通知する場合、EventBridgeやUser NotificationsをChatbotと組み合わせて利用することで、ノーコードで通知を実装することができます。

見た目も下記のようになり良い感じです。

この構成は下記ドキュメントの手順で実装することができます。

課題

先述したノーコードで実装できる通知は便利なのですが、「アカウント名が表示されない」点がわたしたちにとっては課題だと感じていました。
下図のとおりアカウント番号は表示されますが、アカウント名は表示されません。

なぜアカウント名が知りたいのか？

弊社では複数のAWSアカウントをOrganizationsで管理しています。
AWSの全体管理者としてOrganizationsに所属する全アカウントの通知を受け取っていたため、

アカウントが多いとアカウント番号だけでは何のシステムか判別できない

この点について、なんとかならないものかと感じていました。
通知の度にアカウント番号からシステム名を調べるのはなかなかしんどいです。

AWSサポート回答

マネージドな機能で実現できるのが一番ありがたいですが、AWSサポートに問い合わせたところ、残念ながら現状（2024/9/4時点）は実現不可とのことで、機能要望としてお願いさせていただきました。

ドキュメントを見てみた

Healthイベントの仕様はドキュメントで公開されています。

アカウント番号からアカウント名を取得する処理を加えるだけで済みそうであったため、カスタマイズした通知の仕組みを実装してみることにしました。

カスタマイズしてみた

カスタマイズ内容

独自に追加した機能は下記4点です。

アカウント名を表示
日本語に翻訳
イベントの開始終了時刻を日本時間で表示
絵文字をパッと見でわかりやすいものに変更

せっかくなのでアカウント名の表示以外のカスタマイズもしてみました。
実際の通知画面はこのような感じになりました。

一番の目的だったアカウント名にモザイクがかかっていますが...
英語でアカウント名が表示されています。

構成

構成は下図のとおりです。

組織ビュー(ドキュメント)を利用することで、Organizationsに所属する全アカウントのイベントを一か所で確認できるようにしています。
組織ビューの管理をOrganizationsのメンバーアカウントに委任(ドキュメント)することで、管理アカウントの責務を減らしています。
Lambdaでカスタマイズ処理を行います。
Translateで英語の通知内容を日本語に翻訳します。
同一イベントの更新情報をSlack内でスレッド化するために、DynamoDBにイベント情報を保持します。
イベント情報を永続的に保持する必要はないため、DynamoDBに登録するアイテムにTTL属性を設定して自動削除されるようにします。

実装のポイント

Slack連携

Slackとの連携にはBolt for Pythonというフレームワークを利用しました。
Lambda上で容易に利用できる仕組みが整っており大変便利です。

アカウント名の取得

Healthイベントにはアカウント番号が含まれているため、Organizationsのメソッドを利用することでアカウント名が取得できます。

organizations = boto3.client("organizations")

def get_account_name(account_id: str) -> str:
    """
    アカウントIDからアカウント名を取得する
    """
    response = organizations.describe_account(
        AccountId=account_id
    )
    return response["Account"]["Name"]

日本語に翻訳

Amazon Translateを利用しました。

translate = boto3.client('translate')

def translate_text(text: str) -> str:
    """
    英語を日本語に翻訳する
    """
    response = translate.translate_text(
        Text=text,
        SourceLanguageCode="en",
        TargetLanguageCode="ja"
    )
    return response["TranslatedText"]

注意点としては、英語原文と日本語翻訳文が含まれている場合があり、日本語をTranslateで翻訳すると摩訶不思議な文章になってしまいます。
最近あまり見かけなくなってきましたが、AWSの翻訳ドキュメントを思い出しました…

そこでEnglish follows Japaneseという文章が含まれている場合は翻訳しないようにしました。

    # 説明文を翻訳する
    match language:
        case "en_US":
            if "English follows Japanese" not in description:
                description = translate_text(description)
                is_translated = True
        case "ja_JP":
            pass
        case _:
            logger.error(f"unexpected language: {language}")

イベントの開始終了時刻を日本時間で表示

イベント発生時刻はSlackのメッセージ投稿の時刻から判断することもできますが、できれば日本時間で確認したいなと思いました。

def format_datetime(datetime_str: str) -> str:
    """
    HealthEventの時刻フォーマットをISO8601形式の日本時間に変換する
    """
    dt = datetime.strptime(datetime_str, "%a, %d %b %Y %H:%M:%S %Z")
    dt_jst = dt.astimezone(ZoneInfo("Asia/Tokyo"))
    iso_format = dt_jst.isoformat()
    return iso_format

絵文字をパッと見でわかりやすいものに変更・通知先Slackチャンネル分割

視覚的に判別しやすくするために、イベントのステータスごとに絵文字を変えています。
標準的な絵文字のコードはこちらから確認できます。
また、最低限の仕分けとしてイベントカテゴリ(issue accountNotification investigation scheduledChange)ごとにSlackのチャンネルを分けました。

match event_type_category:
    case "issue":
        channel = SLACK_CHANNEL_ID_ISSUE
        match status_code:
            case "open":
                status_icon = ":exclamation:"
            case "closed":
                status_icon = ":white_check_mark:"
            case "upcoming":
                status_icon = ":warning:"
            case _:
                logger.error(f"unexpected status code: {status_code}")
                return
    case "accountNotification":
        channel = SLACK_CHANNEL_ID_ACCOUNT_NOTIFICATION
        status_icon = ":mega:"
    case "scheduledChange":
        channel = SLACK_CHANNEL_ID_SCHEDULED_CHANGE
        status_icon = ":mega:"
    case "investigation":
        channel = SLACK_CHANNEL_ID_INVESTIGATION
        status_icon = ":mega:"
    case _:
        logger.error(f"unexpected event type category: {event_type_category}")
        return

デプロイ方法

Healthイベントを受け取るにあたり、どのようなタイプのイベントをどこのリージョンで受け取れるかを考慮する必要があります。

EventBridge ルールは、AWS Health イベントを受信するリージョンごとに作成する必要があります。ルールを作成しなければ、イベントは受信されません。例えば、米国西部 (オレゴン) リージョンからイベントを受信するには、このリージョンのルールを作成する必要があります。

AWS Health イベントにはリージョン固有ではないものもあります。リージョンに固有ではないイベントはグローバルイベントと呼ばれます。これには、AWS Identity and Access Management (IAM) について送信されるイベントが含まれます。グローバルイベントを受信するには、米国東部 (バージニア北部) リージョンをプライマリーリージョンとし、米国西部 (オレゴン) をバックアップリージョンにするルールを作成する必要があります。

つまり、日本の企業で東京リージョンをメインに利用している場合、少なくともバージニアと東京にLambda等の各リソースをデプロイする必要があります。
今回はデプロイ手段としてAWS CDK(ドキュメント)を利用しました。

Lambdaコード全量

app.handlerがLambda関数のハンドラーです。

app.py

import json
import os
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from utils import (
    get_ssm_parameter,
    translate_text,
    format_datetime,
    get_account_name,
    put_item,
    get_thread_ts,
    is_excluded_event_type_code
)

logger = logging.getLogger(__name__)
logger.setLevel("INFO")

SLACK_BOT_TOKEN_SSM_PARAM_KEY = os.environ["SLACK_BOT_TOKEN_SSM_PARAM_KEY"]
SLACK_CHANNEL_ID_ISSUE = os.environ["SLACK_CHANNEL_ID_ISSUE"]
SLACK_CHANNEL_ID_ACCOUNT_NOTIFICATION = os.environ["SLACK_CHANNEL_ID_ACCOUNT_NOTIFICATION"]
SLACK_CHANNEL_ID_SCHEDULED_CHANGE = os.environ["SLACK_CHANNEL_ID_SCHEDULED_CHANGE"]
SLACK_CHANNEL_ID_INVESTIGATION = os.environ["SLACK_CHANNEL_ID_INVESTIGATION"]

web_client = WebClient(token=get_ssm_parameter(SLACK_BOT_TOKEN_SSM_PARAM_KEY, with_decryption=True))

def handler(event, context): 
    logger.info(f"event: {json.dumps(event,ensure_ascii=False)}")

    detail = event["detail"]

    # ページ分割が発生した場合、2ページ目以降は通知しない
    if detail["page"] != "1":
        logger.info("process ended because page number is not 1")
        return

    # 必須パラメータの取得
    affected_account_id = detail["affectedAccount"]
    event_description = detail["eventDescription"]
    event_type_category = detail["eventTypeCategory"]
    status_code = detail["statusCode"]
    event_arn = detail["eventArn"]
    event_region = detail["eventRegion"]
    event_type_code = detail["eventTypeCode"]
    service = detail["service"]
    event_scope_code = detail["eventScopeCode"]
    start_time = format_datetime(detail["startTime"])
    total_pages = detail["totalPages"]

    language = event_description[0]["language"]
    description = event_description[0]["latestDescription"]

    # 通知除外対象のイベントタイプをチェック
    if is_excluded_event_type_code(event_type_code):
        logger.info(f"{event_type_code} is excluded")
        return

    # 任意パラメータの取得
    affected_entities = detail.get("affectedEntities")
    end_time = format_datetime(detail["endTime"]) if "endTime" in detail else "-"
    
    # フラグ初期値
    is_single_thread = True
    is_translated = False

    # 説明文を翻訳する
    match language:
        case "en_US":
            if "English follows Japanese" not in description:
                description = translate_text(description)
                is_translated = True
        case "ja_JP":
            pass
        case _:
            logger.error(f"unexpected language: {language}")

    # sectionブロックtextフィールドの文字数上限3,000に引っかからないよう切り詰める
    max_length = 2990
    if len(description) > max_length:
        description = description[:max_length] + "..."

    # アカウント名を取得する
    affected_account_name = get_account_name(affected_account_id)
    
    # イベント種別ごとにチャンネルとステータスアイコンをセット
    match event_type_category:
        case "issue":
            channel = SLACK_CHANNEL_ID_ISSUE
            match status_code:
                case "open":
                    status_icon = ":exclamation:"
                case "closed":
                    status_icon = ":white_check_mark:"
                case "upcoming":
                    status_icon = ":warning:"
                case _:
                    logger.error(f"unexpected status code: {status_code}")
                    return
        case "accountNotification":
            channel = SLACK_CHANNEL_ID_ACCOUNT_NOTIFICATION
            status_icon = ":mega:"
        case "scheduledChange":
            channel = SLACK_CHANNEL_ID_SCHEDULED_CHANGE
            status_icon = ":mega:"
        case "investigation":
            channel = SLACK_CHANNEL_ID_INVESTIGATION
            status_icon = ":mega:"
        case _:
            logger.error(f"unexpected event type category: {event_type_category}")
            return
    
    console_link = f"https://health.console.aws.amazon.com/health/home?#/organization/event-log?eventID={event_arn}&eventTab=details"

    blocks = [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*<{console_link}|{status_icon} AWS Health Events | {event_region} | {affected_account_name}>*"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": description
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Affected Account*\n{affected_account_id}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Event Region*\n{event_region}"
                    }
                ]
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Service*\n{service}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Event Type Code*\n{event_type_code}"
                    }
                ]
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Event Type Category*\n{event_type_category}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Event Scope Code*\n{event_scope_code}"
                    }
                ]
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Start Time*\n{start_time}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*End Time*\n{end_time}"
                    }
                ]
            }
        ]
    
    if affected_entities:        
        entity_values = "\n".join(entity.get("entityValue", "") for entity in affected_entities)

        if total_pages != "1":
            # ページ分割されている場合
            entity_values += "\n...and more"
        
        blocks.append({
            "type": "section",
            "fields": [
                {
                    "type": "mrkdwn",
                    "text": f"*Affected Entities*\n{entity_values}"
                }                        
            ]
        })

    # 翻訳文の場合はその旨を明示する
    if is_translated:
        blocks.append({
            "type": "divider"
        })
        blocks.append({
            "type": "context",
            "elements": [
                {
                    "type": "mrkdwn",
                    "text": "Amazon Translateによる翻訳文です"
                }
            ]
        })

    params = {
        "channel": channel,
        "text": "AWS Health Events",
        "blocks": blocks
    }

    # 同一イベントの続報はスレッドで投稿、かつチャンネルにも投稿する
    thread_ts = get_thread_ts(event_arn)
    if thread_ts is not None:
        params["thread_ts"] = thread_ts
        params["reply_broadcast"] = True
        is_single_thread = False

    try:
        response = web_client.chat_postMessage(**params)
        logger.info(f"slack response: {response}")

        if is_single_thread:
            # 初回通知の場合はDynamoDBに登録
            put_item(event_arn, response["message"]["ts"])
        
    except SlackApiError as e:
        logger.exception(e.response["error"])

    return

utils.py

import json
import os
import logging
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from typing import Union
import boto3

logger = logging.getLogger(__name__)

ssm = boto3.client('ssm')
translate = boto3.client('translate')
organizations = boto3.client("organizations")
dynamodb = boto3.client("dynamodb")

DYNAMODB_TABLE_NAME = os.environ["DYNAMODB_TABLE_NAME"]
EXCLUSION_LIST_PARAM_KEY = os.environ["EXCLUSION_LIST_PARAM_KEY"]

def get_ssm_parameter(key: str, with_decryption: bool = False) -> str:
    """
    SSMパラメータ値を取得する
    """
    response = ssm.get_parameter(
        Name=key,
        WithDecryption=with_decryption
    )
    return response["Parameter"]["Value"]

def translate_text(text: str) -> str:
    """
    英語を日本語に翻訳する
    """
    response = translate.translate_text(
        Text=text,
        SourceLanguageCode="en",
        TargetLanguageCode="ja"
    )
    logger.info(f"translate response: {json.dumps(response, ensure_ascii=False)}")
    return response["TranslatedText"]

def format_datetime(datetime_str: str) -> str:
    """
    HealthEventの時刻フォーマットをISO8601形式の日本時間に変換する
    """
    dt = datetime.strptime(datetime_str, "%a, %d %b %Y %H:%M:%S %Z")
    dt_jst = dt.astimezone(ZoneInfo("Asia/Tokyo"))
    iso_format = dt_jst.isoformat()
    return iso_format

def get_account_name(account_id: str) -> str:
    """
    アカウントIDからアカウント名を取得する
    """
    response = organizations.describe_account(
        AccountId=account_id
    )
    return response["Account"]["Name"]

def put_item(eventArn: str, thread_ts: str) -> None:
    """
    DynamoDBテーブルにイベント情報を登録する
    """
    response = dynamodb.put_item(
        TableName = DYNAMODB_TABLE_NAME,
        Item={
            "eventArn": {
                "S": eventArn,
            },
            "thread_ts": {
                "S": thread_ts,
            },
            "ttl": {
                "N": str(int((datetime.now() + timedelta(days=90)).timestamp()))
            },
        }
    )
    logger.info(f"put item response: {json.dumps(response, ensure_ascii=False)}")

def get_thread_ts(eventArn: str) -> Union[str, None]:
    """
    DynamoDBテーブルからイベント情報を取得する
    """
    response = dynamodb.get_item(
        TableName = DYNAMODB_TABLE_NAME,
        Key={
            "eventArn": {
                "S": eventArn,
            }
        }
    )
    if "Item" in response:
        return response["Item"]["thread_ts"]["S"]
    else:
        return None

def is_excluded_event_type_code(event_type_code: str) -> bool:
    """
    通知対象のイベントかどうかを判定する
    """
    exclusion_list_str = get_ssm_parameter(EXCLUSION_LIST_PARAM_KEY)
    exclusion_list = [x.strip() for x in exclusion_list_str.split(',')]
    if event_type_code in exclusion_list:
        return True
    else:
        return False

改善点

今回のカスタマイズ実装は同じような通知の集約に対応できていません。

User Notificationsの場合、通知が大量に届かないよう、指定期間のHealthイベントを集約して通知することができます。
選択肢としては「5分以内に受領」「12時間以内に受領」「集約しない」があります。

通知対象アカウントが多ければ多いほど同様の通知がアカウントの数だけ届くことになるため、ある程度集約されると大量の通知でSlackチャンネルが埋め尽くされることがなくなるメリットがあります。
こちらについては今後必要に応じて追加で実装できればと考えています。

さいごに

今回、マネージドで提供されていない機能を追加で実装してみました。
ただ、追加部分の実装は容易でも、元々マネージド機能として提供されていた部分まで自前で実装する必要はあり、「車輪の最発明」の要素が多分にありました。

しかしそれでも個人的には下記の点でトライしてみて良かったと感じています。

Lambda, Python, boto3, Slack等について学ぶことができた。
マネージドサービスを組み合わせることでプラスαの要素を思っていたよりも容易に加えられることがわかった。
HealthイベントのSlack表示を自由にカスタマイズできるようになった。

全アカウントの通知を一元的にチャットツール上で確認する運用は、そもそもニッチなユースケースなのかもしれませんが…
どなたかの参考になれば幸いです。