GoogleCloudの予算アラートをTeamsへ飛ばしたい・・！

はじめに

こんにちは、博報堂テクノロジーズ・インフラ開発一部の近藤です。
とある内製案件で、Google Cloudの予算アラートをTeamsへ通知したいという要望があり、検証してみましたので、ご紹介です。
単に予算アラート→Teamsへ通知、という単純な感じでは実装が難しかったため、自身の備忘も含めの投稿です。

当内容は2025/5/30時点のモノです。
Google Cloud等の仕様変更に伴い実装処理は変わりますのでご留意ください。
実装内容は個人で検証した一例のため、あくまで参考として捉えてください。

そもそもなぜTeams通知？

Google Cloudの予算アラートの通知先はメールやslack等、様々実装できます。
所属している組織や部署のポリシーとしてslack利用が禁止、メールだと埋もれてしまうのでチャットが良い、というケースがあると思います。
今回は内製プロジェクトで普段使用している連絡手段がTeamsということもあり、Teams連携を採用しています。
なお、個人的な感覚ですが、メールやslack連携が手っ取り早く実装もラクなため、メールに方式を倒しても良いかなと思いましたが、手段として選択肢を持っておくのは大事だと思い、今回の実装に至ってます。

アーキテクチャ

ざっくりこんな感じです。

Google Cloudの世界

Cloud Billing

予算アラートを設定します。通知先としてpub/subトピックを指定します。
なお、プロジェクトが組織に所属しており、特定の組織ポリシーを有効にしている場合、予算アラートの設定ができません。詳細は後述します。
また、Google Cloudの仕様上、予算アラートは閾値超過後に数回アラートを発砲するようです。検証時は、2～3通/1h来ました。詳細は以下を参照ください。

Pub/Sub

予算アラートとCloud Run Functionsを繋ぎます。特殊な設定はありません。

Cloud Run Functions

受け取った予算アラートのメッセージをwebhook経由でTeamsへ連携する処理を実装します。なお、Cloud Billingで記載したアラートの数回発砲の頻度をコントロールする処理も含みます。

Cloud Storage

Cloud Run Functionsの発火時刻を保存します。Race condition回避のためにDB等を使うのも手ですが、そもそもCloud Run Functionsの実行間隔はミリミリでは無いので不要と判断しました。このため、簡易実装を目的にGCSを利用します。

IAM

Cloud Run FunctionsとCloud Run FunctionsのTrigger用のサービスアカウントを作成します。付与するロールは後述します。

Teamsの世界

Workflows

Webhook利用のためにWorkflows(Teamsのチャネルの機能)を利用します。

チャネル

予算アラートを受け取り表示します。

アウトプットイメージ

Teamsチャネル上ではこんな感じの通知が飛びます。(一部情報をマスクしています)

個人的には割と良い感じに飛ばしてくれるなぁと思いました。

ここからは作りの話です。リソースの作成順でご説明します。

環境構築(Teams)

Workflow設定

通知を飛ばしたいチャネルのWebhook URLを取得します。
チャネルの3点リーダーからワークフローを選択します。

webhook要求を受信するとチャネルに投稿するを選択し、次へを押下します。

通知を飛ばしたい対象のチームとチャネルを選択(自動入力なはずです)し、フローの作成を押下します。

出力されるwebhook用のURLを保管しておいてください。

環境構築(Google Cloud)

以降は予算アラートの各種リソースを作成するプロジェクトで作業します。
なお、以降はコンソールを想定した作業です。

IAMの作成

以下条件でサービスアカウントを2個作成します。

Cloud Run Functions本体用のサービスアカウント

アタッチするロール

• roles/run.servicesInvoker(Cloud Run サービス起動元)
• roles/eventarc.eventReceiver(Eventarcイベント受信者)

Cloud Run Functions Trigger用のサービスアカウント

アタッチするロール

• roles/storage.objectAdmin(Storage オブジェクト管理者)

pub/sub

デフォルトの条件でトピックを作成ください。
なお、サブスクリプションは不要です。

予算アラートの作成

3.操作以外は任意の条件で予算アラートを作成ください。
3.操作では以下の通り、先の手順で作成したpub/subトピックを選択ください。

補足1：組織ポリシーに伴いエラーとなるケース

対象のプロジェクトが組織に所属しており、以下の組織ポリシーが有効となっている場合、予算アラートの作成に失敗しました。

• 組織ポリシー名：iam.allowedPolicyMemberDomains
• 

回避方法ですが、プロジェクトレベルのスコープで、この組織ポリシー(共有先のドメインを制限)をGoogle で管理されるデフォルト値にすればOKです。予算アラート設定後は、この設定を戻すのを忘れずに！

これは私の想像ですが、
予算アラート作成時、選択したpub/subトピックに対し、システム管理のサービスアカウント([e-mail])に対してPub/Sub パブリッシャーの権限を付与しますが、この権限付与元はシステムで管理されたプロジェクト等になるため、そのCustomerIDを許可していないからエラーとなるのでは？と考えています。

補足2：メール通知は必須

通知の管理のメール通知アラートを課金管理者とユーザーに送信するにチェックが入ってますが、これはGoogle Cloudの仕様上外せません。外すと以下のエラーになります。

GCSバケットの作成

パブリック非公開でバケットを作成します。その他の項目は任意でOKです。

Cloud run functionsの作成

ここが本丸なので、詳しく説明します。
以下条件でCloud Run Functionsを作成します。

• サービス名：任意
• リージョン：asia-northeast1
• ランタイム：python3.12
• トリガー
  • 種別：pub/subトリガー。上で作成したpub/subトピックを選択。
  • リージョン：asia-northeast1
  • サービスアカウント：IAMの作成で作成したCloud Run Functions Trigger用のサービスアカウントを指定
    補足：Cloud Pub/Sub で ID トークンを作成するには・・・という警告が出力され
    　　　SAに権限付与を推奨されるが、当該権限は不要なので無視でOK。
• 認証：認証が必要を選択
• コンテナの設定
  • 環境変数
    • TEAMS_WEBHOOK_URL
      説明　：TeamsのWorkflow設定で払い出したwebhook用のURL。
      入力例： https://prod- ・・・
    • NOTIFY_STATE_BUCKET
      説明　：GCSバケットの作成で作成したバケット名
      入力例： kondo-test-bucket
    • PROJECT_ID
      説明　：予算の監視をしたいプロジェクトID。
      　　　　複数プロジェクトの場合は特定の一つを入力。
      　　　　アラート内の「GCP予算ページへ移動」のリンクに埋め込む箇所で使用。
      入力例： kondo-test-pj
    • NOTIFY_INTERVAL
      説明　：通知間隔(分)。1日1回通知であれば1440を指定。
      入力例： 1440
  • セキュリティ
    　IAMの作成で作成したCloud Run Functions本体用のサービスアカウントを指定。

上記設定で作成します。作成完了後、以下のコードをデプロイします。
なお、環境固有の情報はコンテナの設定の環境変数に落としているので、基本は以下コードをそのままコピー＆ペーストするだけでOKです。

main.py

main.py

import base64
import json
import logging
import os
from datetime import datetime
from datetime import timedelta

import functions_framework
import requests
from google.cloud import storage

# 環境変数
TEAMS_WEBHOOK_URL = os.environ.get("TEAMS_WEBHOOK_URL")
NOTIFY_STATE_BUCKET = os.environ.get("NOTIFY_STATE_BUCKET")
STATE_FILE = "alertitme.json"
PROJECT_ID = os.environ.get("PROJECT_ID")
NOTIFY_INTERVAL = int(os.environ.get("NOTIFY_INTERVAL"))

# ロガーの設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@functions_framework.cloud_event
def budget_alert_to_teams(cloud_event):
    """
    Pub/Subメッセージでトリガーされ、予算アラートをTeamsに通知するCloud Functions。
    Args:
        cloud_event (cloudevents.http.CloudEvent): CloudEventオブジェクト
    """
    if not TEAMS_WEBHOOK_URL:
        logger.error("TEAMS_WEBHOOK_URLが設定されていません。")
        return "TEAMS_WEBHOOK_URL not set", 500

    try:
        # Pub/Subメッセージの属性情報を取得。属性情報は非エンコード。
        pubsub_message_attribute = cloud_event.data["message"]["attributes"]
        # Pub/Subメッセージのデータをデコード
        pubsub_message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode("utf-8")
        # 受信したpub/subメッセージを出しとく
        logger.info(f"受信したPub/Subメッセージ: {pubsub_message_data}")
        budget_attributes = pubsub_message_attribute
        budget_data = json.loads(pubsub_message_data)

        # 予算アラートの詳細情報を取得
        budget_display_name = budget_data.get("budgetDisplayName", "N/A")
        cost_amount = budget_data.get("costAmount", 0.0)
        cost_interval_start = budget_data.get("costIntervalStart", "N/A")
        budget_amount = budget_data.get("budgetAmount", 0.0)
        currency_code = budget_data.get("currencyCode", "")
        alert_threshold_exceeded = budget_data.get("alertThresholdExceeded", "N/A")
        billing_account_id = budget_attributes.get("billingAccountId", "N/A")  # 同上

        # Teamsに送信するメッセージを作成
        message_title = f"🚨 GCP予算アラート: {budget_display_name}"
        message_text = (
            f"**予算名:** {budget_display_name}\n\n"
            f"**現在の費用:** {cost_amount} {currency_code}\n\n"
            f"**予算額:** {budget_amount} {currency_code}\n\n"
            f"**超過したしきい値:** {alert_threshold_exceeded * 100 if isinstance(alert_threshold_exceeded, float) else 'N/A'}%\n\n"
            f"**予算のしきい値監視開始期間:** {cost_interval_start}\n\n"
            f"[GCP予算ページへ移動](https://console.cloud.google.com/billing/{billing_account_id.replace('billingAccounts/', '')}/reports;projects={PROJECT_ID}?project={PROJECT_ID})"
        )

        teams_payload = {
            "type": "message",
            "attachments": [
                {
                    "contentType": "application/vnd.microsoft.card.adaptive",
                    "contentUrl": None,
                    "content": {
                        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                        "type": "AdaptiveCard",
                        "version": "1.5",
                        "body": [
                            {"type": "TextBlock", "text": message_title, "weight": "Bolder", "size": "Medium"},
                            {"type": "TextBlock", "text": message_text, "wrap": True},
                        ],
                    },
                }
            ],
        }
        # 通知頻度制御（過去通知時刻を確認）
        storage_client = storage.Client()
        bucket = storage_client.bucket(NOTIFY_STATE_BUCKET)
        blob = bucket.blob(STATE_FILE)
        now = datetime.utcnow()
        notify = True

        if blob.exists():
            state = json.loads(blob.download_as_text())
            last_time = datetime.fromisoformat(state.get("last_notify_time"))
            # NOTIFY_INTERVALより短い時間ならば通知しない
            if now - last_time < timedelta(minutes=NOTIFY_INTERVAL):
                notify = False

        if notify:
            # Teamsに通知を送信
            response = requests.post(TEAMS_WEBHOOK_URL, json=teams_payload, timeout=10)
            response.raise_for_status()  # ステータスコードが2xxでない場合に例外を発生
            # 通知記録の更新
            blob.upload_from_string(json.dumps({"last_notify_time": now.isoformat()}))

            logger.info("Teamsへの通知に成功しました。")

            return "OK", 200
        else:
            print("Skipped notification due to frequency control.")

    except requests.exceptions.RequestException as e:
        logger.error(f"Teamsへの通知中にエラーが発生しました (RequestException): {e}")
        return f"Error sending to Teams: {e}", 500
    except json.JSONDecodeError as e:
        logger.error(f"Pub/SubメッセージのJSONデコードに失敗しました: {e}")
        logger.error(f"失敗したデータ: {cloud_event.data['message']['data']}")
        return "Error decoding Pub/Sub message", 400
    except KeyError as e:
        logger.error(f"Pub/Subメッセージの必須キーが見つかりません: {e}")
        logger.error(f"受信データ: {budget_data if 'budget_data' in locals() else 'N/A'}")
        return f"Missing key in Pub/Sub message: {e}", 400
    except Exception as e:
        logger.error(f"予期せぬエラーが発生しました: {e}")
        return "Internal Server Error", 500

requirements.txt

requirements.txt

requests
google-cloud-storage

デプロイ前に、関数のエントリ ポイントをbudget_alert_to_teamsへ変更ください

アウトプットイメージ(再掲)

上記設定が上手くいっていれば、Teamsチャネル上ではこんな感じの通知が飛びます。(再掲)

運用

この処理をプロダクションで稼働させる場合、運用を考慮する必要があります。
当処理での運用の要素は以下２点と考えています。

• 予算アラートの閾値見直し
• Cloud Run Functionsのラインタイム

予算アラートの閾値見直し

Google Cloudの利用状況が可変な場合、形骸化しないような予算アラートを設定すべきです。例えば、予算アラートを受け取ったとしても、超えて当然として何も対処しないことが常態化している場合、予算アラートの意味合いがなくなっています。
利用料の推移を観測し、適切な予算アラートの閾値となるよう定期的に見直すべきだと考えています。

Cloud Run Functionsのラインタイム

今回はPython3.12を利用しましたが、EOSに応じてバージョンアップが必要となります。
以下などのドキュメントを参照し、適切なタイミングでバージョンアップを検討し実装すべきだと考えています。
（個人的には、バージョンアップ後のトラブルに余裕を持つため、非推奨日より１、２ヶ月ほど前に実施すべきだと考えています。）

そもそもなぜこの記事を書こうと思ったか？

私がググった限りでは、予算アラートのslack連携のページはヒットしましたが、Teams連携を紹介しているモノがあまりなく、今回、トラブルがありつつも諸々調べながら実装したため、ノウハウとしてまとめておこうと思った次第です。

最後に

今後も積極的に実際のコードや設定手順を通じて、具体的なノウハウを共有していきたいと考えています！