概要
Cloud Functionsのエラー内容をSlackに通知したい。
ログルーターを使用した方法が簡単だったので作り方をまとめておく。
Cloud Functionsの使用ランタイムはpython3.7
処理の流れとしては下記になる。
- 実行関数のエラーをCloud Loggingが検知
- 予め作成したログルーターのシンク設定に従って、指定のpub/subトピックを送信
- pub/subトピックをトリガーにしたslackへの通知用関数を実行
- slackにメッセージが表示される
pub/subトピック作成
コンソール画面からトピックを作成する。
ログルーターのシンク作成
Cloud Loggingの「ログルーター」の項目で新規にシンクを作成。
どういった時にルーターがシンクするかはCloud Loggingのクエリ言語で細かく指定することができる。
今回はシンクする最低条件として「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
}))