GCP予算アラートをChatworkに通知する方法｜Pub/SubとCloud Run Functions(第2世代)で自動化

GCP（Google Cloud）を利用していて一番怖いのは「いつの間にか予算を超えていた」という事態ですよね。標準のメール通知だと見逃しがちですが、普段使っているChatworkやLINEに通知が来れば安心です。今回は、Pub/SubとCloud Run Functions（第2世代）を組み合わせて、予算状況を自動通知する仕組みを構築した際の備忘録をまとめます。

1. 全体の構成図

通知の流れは以下の通りです。

Cloud Billing: 予算のしきい値を超えたらメッセージを発行
Pub/Sub: メッセージを受け取り、後続の関数へ渡す
Cloud Run Functions: Pythonでメッセージを加工し、Chatwork APIを叩く

2. 事前準備

通知先となるChatworkで以下の情報を取得しておきます。

APIトークン
通知先のルームID

3. GCP側の設定手順

① Pub/Subトピックの作成

まず、予算アラートの中継点となるトピックを作成します。

GCPコンソールで「Pub/Sub」→「トピック」に移動し、任意のID（例：budget-alerts）で作成します。

② 予算アラートの設定

「お支払い」→「予算とアラート」から、対象の予算設定を開きます。
「アクション」セクションで、**「Pub/Sub トピックをこの予算に接続する」**にチェックを入れ、先ほど作ったトピックを選択します。

③ Cloud Run Functionsの作成

第2世代（Python 3.11以上推奨）で作成します。

トリガー: Cloud Pub/Sub
イベントの種類: google.cloud.pubsub.topic.v1.messagePublished
認証: 「認証が必要」を選択
Ingress: 「すべてのトラフィックを許可」

4. 実装コード（Python）

ここが一番のハマりポイントでした。第2世代では引数が (cloud_event) の1つになります。また、初期化時に飛んでくる「0円」の通知を無視するロジックを入れています。

main.py

Python

import base64
import json
import requests

def notify_chatwork(cloud_event):
    try:
        # データのデコード
        data_wrapper = cloud_event.data
        if isinstance(data_wrapper, (str, bytes)):
            data_wrapper = json.loads(data_wrapper)
        
        pubsub_data = data_wrapper["message"]["data"]
        message_str = base64.b64decode(pubsub_data).decode("utf-8")
        budget_info = json.loads(message_str)
        
        # データの抽出
        cost = budget_info.get('costAmount', 0)
        budget = budget_info.get('budgetAmount', 0)
        currency = budget_info.get('currencyCode', 'JPY')

        # 利用額が 0 の場合はノイズになるのでスキップ
        if cost <= 0:
            return "Skipped: Cost is 0", 200
        
        # 通知メッセージ作成
        message = f"【GCP予算アラート】\n現在の利用額: {cost} {currency}\n予算設定額: {budget} {currency}"
        
        # Chatwork送信
        url = "https://api.chatwork.com/v2/rooms/【ルームID】/messages"
        headers = {"X-ChatWorkToken": "【APIトークン】"}
        params = {"body": message}
        
        res = requests.post(url, headers=headers, data=params)
        res.raise_for_status()
        
        return "OK", 200

    except Exception as e:
        print(f"Error detail: {e}")
        return f"Error: {e}", 500

requirements.txt

Plaintext

functions-framework==3.*
requests==2.31.0

5. ハマりポイントと解決策

構築中に遭遇したエラーとその対策です。

403 Forbiddenエラー: Pub/Subサービスエージェントに「Cloud Run 起動元」の権限を付与する必要があります。IAM画面で「Google提供のロール付与を含める」にチェックを入れて設定しましょう。
NameError: name ‘event’ is not defined: 第1世代のサンプルコードを参考にすると引数名でエラーになります。第2世代では cloud_event を使いましょう。
通知が大量に来る: 予算設定直後などに0円の通知が複数来ることがあります。コード内で cost <= 0 の判定を入れることで解決しました。

まとめ

これでGCPの予算をリアルタイムに近い形で監視できるようになりました！一度設定してしまえば、他のプロジェクトでも応用が効くのでおすすめです。