[GCP]Cloud Functionsのエラーをslackに通知する

概要

Cloud Functionsのエラー内容をSlackに通知したい。
ログルーターを使用した方法が簡単だったので作り方をまとめておく。
Cloud Functionsの使用ランタイムはpython3.7

処理の流れとしては下記になる。

  • 実行関数のエラーをCloud Loggingが検知
  • 予め作成したログルーターのシンク設定に従って、指定のpub/subトピックを送信
  • pub/subトピックをトリガーにしたslackへの通知用関数を実行
  • slackにメッセージが表示される

pub/subトピック作成

コンソール画面からトピックを作成する。

ログルーターのシンク作成

Cloud Loggingの「ログルーター」の項目で新規にシンクを作成。
どういった時にルーターがシンクするかはCloud Loggingのクエリ言語で細かく指定することができる。

Logging のクエリ言語

LogEntry

今回はシンクする最低条件として「ERROR」ステータスを指定した。

resource.type="cloud_function"
resource.labels.function_name="hoge_func"
severity>=ERROR

pub/subをトリガーにした関数を作成

Cloud Functionsで作成。
slackのwebhook URLをenv.ymlに定数として定義。
pub/subから受け取った情報から関数名やテキスト内容を取得できるので、slackの通知内容に含めた。

import base64
import json
import os
import requests

WEBHOOK_URL = os.environ.get('WEBHOOK_URL')

def notify_alert(event, context):
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    # JSON形式の文字列を辞書に変換
    d = json.loads(pubsub_message)
    post_slack(d)

def post_slack(msg):
    message = '・関数名:\n' + msg['resource']['labels']['function_name'] + "\n" + '・エラー内容:\n' + msg['textPayload']
    requests.post(WEBHOOK_URL, data=json.dumps({
        'text': message
    }))

参考

GCP の Error Reporting に飛ぶエラーを Slack に通知したい

ログルーターの概要

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です

5 × 2 =

このサイトはスパムを低減するために Akismet を使っています。コメントデータの処理方法の詳細はこちらをご覧ください