Zenn
Rehab Tech BlogPublicationへの投稿
🔔

【Google Cloud】Cloud Buildのビルド結果をSlackに通知する

久良木
2025/01/10に公開
Slack
Google Cloud
Cloud Build
tech

はじめに

Rehab for JAPANでCTOを務めている@rkyuragiです。
Rehabではデータ分析基盤やMLOps基盤でGoogle Cloudを活用しており、
各基盤をデプロイするため、Cloud Buildを利用してCI/CDを実現しています。

弊社はコミュケーションツールとしてSlackを利用しているため、
ビルド結果はSlackに通知してビルドのステータスを楽に把握できるようにしたいと考えました。
今回は、Cloud Buildのビルド結果をSlackに通知する方法(Pythonベース)をご紹介します。
ちなみに、今回の方法で構築した場合、特定のCloud Buildだけではなく全てのビルド結果をSlackに通知することができるので、プロジェクトで一度設定しておけばビルドが増えても都度設定する必要がありません。

公式ドキュメントとしては、この辺りが参考になります。
（ただ、初見だとよくわからない。。。）

本記事のターゲット

  • Cloud Buildを利用している方
  • Cloud Buildのビルド結果をSlackに通知する方法を知りたい方

要約

Cloud Buildのビルド結果をSlackに通知するために、
Cloud Buildのビルド結果をPub/Subに通知し、Pub/SubからCloud Run関数を呼び出してSlackに通知する方法を紹介します。

Cloud Buildのビルド結果をSlackに通知する方法

以下の手順でCloud Buildのビルド結果をSlackに通知する方法を説明します。

0. 事前準備

  • Google Cloudのプロジェクトを作成
  • Google CloudのAPIを有効化（抜けがあったらスミマセン）
    • Cloud Build API
    • Cloud Pub/Sub API
    • Cloud Run API
    • Cloud Functions API
    • Secret Manager API
    • IAM API
  • SlackのWebhook URLを取得
  • ローカル環境にgcloudコマンドをインストール(デプロイするため)

1. 各処理の実装

ディレクトリ構成は以下の通りです。

.
├── app
│   ├── main.py
│   └── requirements.txt
├── .env
└── deploy.sh

1-1. メイン処理

./appディレクトリを作成し、
./app/main.pyに以下を記述する。

import base64
import json
import logging
import os
from urllib.parse import urlparse

import requests
from flask import Request

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

# SlackのIncoming Webhook URL
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")

CLOUD_BUILD_STATUS_STATUS_UNKNOWN = "STATUS_UNKNOWN"
CLOUD_BUILD_STATUS_QUEUED = "QUEUED"
CLOUD_BUILD_STATUS_WORKING = "WORKING"
CLOUD_BUILD_STATUS_SUCCESS = "SUCCESS"
CLOUD_BUILD_STATUS_FAILURE = "FAILURE"
CLOUD_BUILD_STATUS_INTERNAL_ERROR = "INTERNAL_ERROR"
CLOUD_BUILD_STATUS_TIMEOUT = "TIMEOUT"
CLOUD_BUILD_STATUS_CANCELLED = "CANCELLED"

STATUS_COLOR_MAP = {
    CLOUD_BUILD_STATUS_STATUS_UNKNOWN: "#FFA500",
    CLOUD_BUILD_STATUS_QUEUED: "#0072EF",
    CLOUD_BUILD_STATUS_WORKING: "#0063A7",
    CLOUD_BUILD_STATUS_SUCCESS: "#00EF7D",
    CLOUD_BUILD_STATUS_FAILURE: "#FF0000",
    CLOUD_BUILD_STATUS_INTERNAL_ERROR: "#FF0000",
    CLOUD_BUILD_STATUS_TIMEOUT: "#FF0000",
    CLOUD_BUILD_STATUS_CANCELLED: "#FF0000",
}

SLACK_MESSAGE_TEMPLATE = """
*Project ID*: {project_id}\n
*Build ID*: {build_id}\n
*TRIGGER Name*: {trigger_name}\n

*GitHub Repository*: <{git_url}|{git_repo_name}>\n

*Logs*: <{log_url}|View Build Logs>
"""


def get_repo_name_from_url(url: str) -> str:
    """GitリポジトリのURLからリポジトリ名を取得する"""
    # URLを解析
    parsed_url = urlparse(url)
    # パスをスラッシュで分割し、最後の要素を取得
    repo_name = parsed_url.path.split("/")[-1]
    # '.git'を削除してリポジトリ名のみ取得
    if repo_name.endswith(".git"):
        repo_name = repo_name[:-4]
    return repo_name


def send_slack_notification(title: str, message: str, color: str = "#FF0000") -> None:
    """Slackに通知を送信する"""
    webhook_url = SLACK_WEBHOOK_URL
    payload = {
        "attachments": [
            {
                "title": title,
                "text": f"{message}",
                "mrkdwn_in": ["text"],
                "color": color,
            }
        ]
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(webhook_url, data=json.dumps(payload), headers=headers)
    if response.status_code != 200:
        raise ValueError(
            f"Request to slack returned an error {response.status_code}, the response is:\n{response.text}"
        )


def app(request: Request) -> str:
    """Pub/Subメッセージを受け取り、Slackに通知するCloud Function"""
    try:
        # Pub/Subメッセージを取得
        envelope = request.get_json()
        if not envelope:
            logging.error("No Pub/Sub message received")
            return ("Bad Request: No Pub/Sub message", 400)

        pubsub_message = envelope.get("message", {}).get("data", "")
        if not pubsub_message:
            logging.error("No data in Pub/Sub message")
            return ("Bad Request: No message data", 400)

        # Base64デコードしてメタデータを取得
        message_data = base64.b64decode(pubsub_message).decode("utf-8")
        build_metadata = json.loads(message_data)

        # Cloud Buildのビルド情報を取得
        status = build_metadata.get("status", "UNKNOWN")
        build_id = build_metadata.get("id", "N/A")
        git_url = (
            build_metadata.get("source", {}).get("gitSource", {}).get("url", "N/A")
        )
        git_repo_name = get_repo_name_from_url(git_url)
        trigger_name = build_metadata.get("substitutions", {}).get(
            "TRIGGER_NAME", "N/A"
        )
        project_id = build_metadata.get("projectId", "N/A")
        log_url = build_metadata.get("logUrl", "N/A")

        if trigger_name == "N/A":
            logging.warning("Trigger name is not available")
            return ("not target.", 200)

        # Slackメッセージを組み立てる
        color = STATUS_COLOR_MAP.get(status, "#FF0000")
        slack_message = SLACK_MESSAGE_TEMPLATE.format(
            project_id=project_id,
            build_id=build_id,
            git_repo_name=git_repo_name,
            git_url=git_url,
            trigger_name=trigger_name,
            log_url=log_url,
        )

        if status == CLOUD_BUILD_STATUS_SUCCESS:
            status = f"✅{status}"
        elif status in (
            CLOUD_BUILD_STATUS_FAILURE,
            CLOUD_BUILD_STATUS_INTERNAL_ERROR,
            CLOUD_BUILD_STATUS_TIMEOUT,
            CLOUD_BUILD_STATUS_CANCELLED,
        ):
            status = f":no_entry:{status}"
            # ビルドが失敗した場合は@hereで通知
            slack_message = "<!here>\n" + slack_message

        title = f"Cloud Build Notification: {status}"

        send_slack_notification(title, slack_message, color)

        logging.info(f"Notification sent for build: {build_id}")
        return ("Success", 200)

    except Exception as e:
        logging.error(f"Error: {str(e)}")
        return ("Internal Server Error", 500)

1-2. 必要ライブラリの設定

./appディレクトリに
./app/requirements.txtに以下を記述する。

Flask==2.3.3
requests==2.25.1
google-cloud-secret-manager==2.7.0

1-3. 環境変数の設定

./.envに以下を記述する。

SLACK_WEBHOOK_URL="https://hooks.slack.com/xxxxxxx

1-4. デプロイスクリプトの作成

./deploy.shに以下を記述する。

source .env

FIRST_DEPLOYMENT=true # 初回デプロイの場合はtrueにする

PROJECT_ID="project_id" # Google CloudプロジェクトID
REGION="us-central1" # デプロイするリージョン

# 関数の設定
FUNCTION_NAME="cloudbuild-notifier"
ENTRY_POINT="app"
RUNTIME="python311" # 使用するランタイムを指定
SOURCE_PATH="./app" # ソースコードのディレクトリ

SERVICE_ACCOUNT_NAME="pubsub-cloud-run-invoker"
SERVICE_ACCOUNT="${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"

gcloud config set project $PROJECT_ID
gcloud config set run/region $REGION

PROJECT_NUMBER=$(gcloud projects describe $(gcloud config get-value project) --format="value(projectNumber)")
echo $PROJECT_NUMBER

if [ $FIRST_DEPLOYMENT = true ]; then
    gcloud pubsub topics create cloud-builds

    printf ${SLACK_WEBHOOK_URL} | gcloud secrets create CLOUD_BUILD_NOTIFIERS_SLACK_WEBHOOK_URL --data-file=-

    gcloud iam service-accounts create ${SERVICE_ACCOUNT_NAME} \
        --display-name "Pub/Sub Cloud Run Invoker"

    gcloud secrets add-iam-policy-binding CLOUD_BUILD_NOTIFIERS_SLACK_WEBHOOK_URL \
        --member="serviceAccount:${SERVICE_ACCOUNT}" \
        --role="roles/secretmanager.secretAccessor"
fi

gcloud functions deploy $FUNCTION_NAME \
    --gen2 \
    --runtime=$RUNTIME \
    --region=$REGION \
    --entry-point=$ENTRY_POINT \
    --source=$SOURCE_PATH \
    --trigger-http \
    --timeout=30s \
    --max-instances 1 \
    --service-account=${SERVICE_ACCOUNT} \
    --set-secrets SLACK_WEBHOOK_URL=projects/${PROJECT_NUMBER}/secrets/CLOUD_BUILD_NOTIFIERS_SLACK_WEBHOOK_URL/versions/latest

if [ $FIRST_DEPLOYMENT = true ]; then
    gcloud pubsub subscriptions create cloud-builds-to-cloud-run \
        --topic=cloud-builds \
        --push-endpoint=https://${REGION}-${PROJECT_ID}.cloudfunctions.net/${FUNCTION_NAME} \
        --push-auth-service-account="${SERVICE_ACCOUNT}"
fi

2. デプロイ

以下のコマンドを実行してデプロイします。

sh deploy.sh

デプロイに成功すると、以下の画像のようにCloud Run関数が作成されます。

CloudRun関数

通知の確認

そして、Cloud Buildのビルドが実行されると、以下の画像のようにSlackにビルド進捗が通知されます。
（通知アイコンはSlack Webhookの設定画面でCloud Buildの画像を登録してます。）

Slack通知
ビルドが開始した場合の通知
Slack通知
ビルドが成功した場合の通知

まとめ

今回は、Cloud Buildのビルド結果をSlackに通知する方法をご紹介しました。
Cloud Buildのビルド結果をPub/Subに通知し、Pub/SubからCloud Run関数を呼び出してSlackに通知することで、
ビルドのステータスをリアルタイムで把握できるようになります。

参考になれば幸いです！

Rehab Tech Blog
Rehab Tech BlogPublication

Rehab for JAPANのTech Blogです。 介護・ヘルスケア関連のサービスを提供しています！

Discussion