GCPのパイプラインが失敗したらSlackに通知する

夜間のバッチが失敗しても、誰も気づかないことがあります。翌朝「昨日のデータが入っていない」と言われて初めて知ります。

原因は監視していないことではなく、失敗が誰にも届いていないことです。管理画面を毎朝見に行く運用は続きません。落ちたときだけ手元に飛んでくる形にします。

問題：失敗が無音で終わる

よくある状態は3つです。

3つ目が一番厄介です。成功は通知しないと決めるだけで、通知の価値が上がります。

仕組み：通知の出口を1本にまとめる

構成は次の通りです。

スケジューラ（時刻で起動）
  ↓
ワークフロー（処理の順番を管理。try で囲む）
  ↓ 失敗したら except へ
通知用のサービス（1本だけ用意する）
  ↓ Secret Manager から Webhook URL を取得
Slack の Incoming Webhook
  ↓
Slackのチャネルにメッセージが届く

要点は、通知を送る処理を1か所にまとめることです。

パイプラインが3本くらいのうちは前者でも回りますが、10本を超えると必ず漏れます。

Slackの Incoming Webhook の性質

先に押さえておく点が2つあります。

つまり通知先を増やすには、追加するチャネル用のWebhookをもう1本発行します。ここを知らずに「送信先の指定を書き換えれば届く」と考えると、丸一日溶けます。

やること

手順1. SlackでWebhookを発行する

1. ブラウザで https://api.slack.com/apps を開き、通知を出したいワークスペースでログインします
2. 「Create New App」→「From scratch」を選びます
3. アプリ名（例：Pipeline Notifier）を入力し、ワークスペースを選んで「Create App」
4. 左メニューの「Incoming Webhooks」を開き、右上のトグルを On にします
5. 下の「Add New Webhook to Workspace」を押します
6. 投稿先のチャネルを選び、「許可する」を押します
7. 表示された https://hooks.slack.com/services/... をコピーします

ワークスペースの設定によっては、アプリの追加に管理者の承認が必要です。その場合は申請してから進みます。

手順2. Webhook URLをSecret Managerに保存する

URLはパスワードと同じです。コードや設定ファイルに直接書かないでください。

1. Google Cloud コンソールで「Secret Manager」を開きます
2. 「シークレットを作成」を押します
3. 名前を付ける（例：slack-webhook）
4. シークレットの値にJSONを貼ります

{
  "webhook_url": "https://hooks.slack.com/services/..."
}

JSONにしておくと、通知先を増やすときにキーを足すだけで済みます。

手順3. 通知用のサービスを用意する

Cloud Run にサービスを1本立てます。役割は「受け取った文言をSlackに投げる」だけです。

処理の中身は次の形にします。

# 登録されているWebhookを全件送信する
webhook_urls = [
    value for key, value in secret.items()
    if key.startswith("webhook_url") and value
]

results = []
for url in webhook_urls:
    try:
        response = requests.post(url, json={"text": text}, timeout=10)
        results.append(response.status_code == 200)
    except Exception as e:
        print(f"[ERROR] slack_notification_failed: {e}")
        results.append(False)

# 1本でも成功すれば成功扱いにする
return any(results)

ポイントはキー名の前方一致で全件に送ることです。こうしておくと、後から通知先を増やすときにSecret Managerへキーを1つ足すだけで済み、コードもパイプライン側も触りません。

もう1つ、1本失敗しても全体を失敗にしません。片方のチャネルの不調で通知そのものが落ちるのは本末転倒です。

手順4. ワークフローの失敗時に呼ぶ

ワークフローの定義で、処理全体を try で囲み、except で通知用サービスを呼びます。

通知の文面には次を必ず入れます。

「エラーが発生しました」だけの通知は、届いても何もできません。

手順5. 動作確認する

通知用サービスに直接リクエストを送り、届くことを確かめます。

curl -X POST "https://<通知用サービスのURL>/message_to_slack" \
  -H "Content-Type: application/json" \
  -d '{"level":"error","message":"接続テスト"}'

確認する項目は次のとおりです。

3つ目を省かないでください。直接呼んで届いても、ワークフローの except の書き方が間違っていて届かないことがあります。

通知先を増やすとき

Secret Managerの値にキーを1つ足します。

{
  "webhook_url": "（既存の値）",
  "webhook_url_dev": "（追加したチャネルのURL）"
}

手順3のコードは前方一致で全件に送るため、これだけで両方に届きます。コードの修正もパイプライン側の修正も不要です。

なお既存のキー名は改名しません。運用中のキー名にtypoがあっても、そのまま使うほうが安全です。改名した瞬間、通知が全部止まります。

失敗例

切り戻し

変更を戻す方法を、作業前に確認しておきます。

Secret Manager は削除ではなく無効化で戻します。削除すると戻せません。

まとめ

• 監視していないことではなく、失敗が誰にも届いていないことが問題
• 通知を送る処理は1本にまとめます。各パイプラインに直接書くと必ず書き漏れます
• Webhook URLはSecret Managerに置きます。キー名の前方一致で全件送信にすると、通知先の追加が設定だけで済みます
• Slackの Incoming Webhook は1本＝1チャネル固定です。増やすにはWebhookを追加発行します
• 通知文にはパイプライン名・実行時刻・環境を入れます。ないと届いても動けません
• 成功は通知しません。流れる通知は見られなくなります

最初にやることはコードを書くことではなく、動いているパイプラインを一覧にして、どれに通知が付いていないかを数えることです。