【Google Cloud】Google Cloudで発生したエラーもAWSのSNSからメール通知したい

import functions_framework
from google.cloud import pubsub_v1
import json
import os
from datetime import datetime, timezone

@functions_framework.http
def error_trigger(request):
    """
    HTTPリクエストを受け取り、意図的にエラー情報をPub/Subに送信する関数
    """
    project_id = os.environ.get('GCP_PROJECT', '{projectID}')
    topic_id = os.environ.get('PUBSUB_TOPIC', 'cm-kawanago-error-topic')

    # Pub/Sub Publisher クライアントを初期化
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path(project_id, topic_id)

    # エラー情報を作成
    error_message = {
        'error_type': 'TestError',
        'message': 'これはマルチクラウドエラー通知のテストです',
        'severity': 'ERROR',
        'source': 'cm-kawanago-error-trigger',
        'timestamp': datetime.now(timezone.utc).isoformat()
    }

    # メッセージをJSON形式にエンコード
    message_data = json.dumps(error_message, ensure_ascii=False).encode('utf-8')

    try:
        # Pub/Subにメッセージを送信
        future = publisher.publish(topic_path, message_data)
        message_id = future.result()

        return {
            'status': 'success',
            'message': 'エラー情報をPub/Subに送信しました',
            'message_id': message_id,
            'topic': topic_path
        }, 200

    except Exception as e:
        return {
            'status': 'error',
            'message': f'Pub/Subへの送信に失敗しました: {str(e)}'
        }, 500

functions-framework==3.*
google-cloud-pubsub==2.*

import functions_framework
import base64
import json
import os
import boto3
import urllib.request
import urllib.parse

def get_google_identity_token():
    """Google Identity Token取得"""
    try:
        metadata_server = 'http://metadata.google.internal/computeMetadata/v1/'
        token_request_url = metadata_server + 'instance/service-accounts/default/identity'
        token_request_headers = {'Metadata-Flavor': 'Google'}

        audience = 'sts.amazonaws.com'
        params = {'audience': audience, 'format': 'full', 'include_email': 'true'}
        url = f"{token_request_url}?{urllib.parse.urlencode(params)}"

        req = urllib.request.Request(url, headers=token_request_headers)
        with urllib.request.urlopen(req) as response:
            identity_token = response.read().decode('utf-8')
        return identity_token
    except Exception as e:
        raise Exception(f"JWT取得エラー: {e}")

@functions_framework.cloud_event
def bridge_to_sns(cloud_event):
    """
    Pub/Subからメッセージを受け取り、JWTを使ってAWS SNSに転送する関数
    """
    # 環境変数から設定を取得
    aws_role_arn = os.environ.get('AWS_ROLE_ARN')
    sns_topic_arn = os.environ.get('SNS_TOPIC_ARN')

    if not aws_role_arn or not sns_topic_arn:
        print("ERROR: AWS_ROLE_ARN または SNS_TOPIC_ARN が設定されていません")
        return

    # Pub/Subメッセージをデコード
    pubsub_message = base64.b64decode(cloud_event.data["message"]["data"]).decode('utf-8')
    print(f"受信したメッセージ: {pubsub_message}")

    try:
        # Google CloudのService AccountからID Tokenを取得
        print("Google CloudのID Tokenを取得中...")
        id_token = get_google_identity_token()
        print("ID Token取得成功")

        # AWS STSクライアントを作成
        sts_client = boto3.client('sts', region_name='us-east-1')

        # AssumeRoleWithWebIdentityでIAMロールを引き受け
        print(f"AWS IAMロール {aws_role_arn} を引き受け中...")
        assume_role_response = sts_client.assume_role_with_web_identity(
            RoleArn=aws_role_arn,
            RoleSessionName='cloud-function-session',
            WebIdentityToken=id_token
        )
        print("IAMロール引き受け成功")

        # 一時認証情報を取得
        credentials = assume_role_response['Credentials']

        # 一時認証情報を使ってSNSクライアントを作成
        sns_client = boto3.client(
            'sns',
            region_name='us-east-1',
            aws_access_key_id=credentials['AccessKeyId'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            aws_session_token=credentials['SessionToken']
        )

        # SNSにメッセージを送信
        print(f"SNSトピック {sns_topic_arn} にメッセージを送信中...")

        # メッセージをパース
        try:
            message_data = json.loads(pubsub_message)
            subject = f"[{message_data.get('severity', 'ERROR')}] {message_data.get('error_type', 'Error')}"
            message_body = json.dumps(message_data, ensure_ascii=False, indent=2)
        except json.JSONDecodeError:
            subject = "Google Cloud エラー通知"
            message_body = pubsub_message

        response = sns_client.publish(
            TopicArn=sns_topic_arn,
            Subject=subject,
            Message=message_body
        )

        message_id = response['MessageId']
        print(f"SNS送信成功: MessageId={message_id}")

    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")
        import traceback
        traceback.print_exc()
        raise

functions-framework==3.*
boto3==1.*

curl -X POST https://asia-northeast1-{projectID}.cloudfunctions.net/cm-kawanago-error-trigger

受信したメッセージ: {"error_type": "TestError", "message": "これはマルチクラウドエラー通知のテストです", "severity": "ERROR", "source": "cm-kawanago-error-trigger", "timestamp": "2025-10-21T02:12:53.193984+00:00"}
Google CloudのID Tokenを取得中...
ID Token取得成功
AWS IAMロール arn:aws:iam::{accountID}:role/cm-kawanago-google-jwt-role を引き受け中...
IAMロール引き受け成功
SNSトピック arn:aws:sns:us-east-1:{accountID}:cm-kawanago-notification にメッセージを送信中...
SNS送信成功: MessageId=dda0975c-1b37-530c-9ce0-f62cf5509e4f

# 件名
[ERROR] TestError

# 本文
{
  "error_type": "TestError",
  "message": "これはマルチクラウドエラー通知のテストです",
  "severity": "ERROR",
  "source": "cm-kawanago-error-trigger",
  "timestamp": "2025-10-21T02:12:53.193984+00:00"
}

{ジョブ名}でエラーが発生しました。

■ 重要度
ERROR

■ エラー内容
これはマルチクラウドエラー通知のテストです

■ 詳細
・エラー種別: TestError
・発生日時: 2025-10-21 02:12:53 (UTC)
・https://ログが確認できるコンソールURL