Cloud MonitoringでSlackに通知する

はじめに

Cloud Monitoringで特定のログを検出した時にSlackに通知する処理を紹介します。
個人用のREST APIをFAST API(Python)で作成しCloud Runでデプロイしたのですが、
自分以外からのアクセスされるのを気にしてちょっとした対策に作りました。

構成図

Cloud RunからのログをCloud Loggingで取得し、特定のログを検出すればCloud MonitoringからSlackに通知(アラート)を送ります。
今回はREST APIのAPIキーが正しくない場合にSlackへ通知を送るようにします。
(REST API上でAPIキーをチェックしていますが、今思い返せばAPI Gatewayの方が良かったかも)

作り方

ログの準備

まず、APIなどで検出するためのログを用意します。
FAST APIの一例です。
今回はAPIキーを作って(かなり)雑なチェック処理になっています。

@app.middleware("https")
async def check_api_key(request: Request, call_next):
    api_key = request.headers.get("X-API-Key")
    if api_key != settings.rest_api_key:
        logging.warning("API access denied")
        raise HTTPException(status_code=401, detail="Invalid API Key")
    response = await call_next(request)
    return response

logging.warning("API access denied")部分を検出する想定です。

Cloud Logging

検出するためのクエリを作成します。

resource.type = "cloud_run_revision"
resource.laよbels.service_name = "cloud_run_service_name"
resource.labels.location = "location"
textPayload: "WARNING:root:API access denied"

クエリから指標を作成します。

指標の作成画面については「ログベースの指標の名前」に名前を記入するくらいで、
他の変更は不要です。

Cloud Monitoring

Cloud Monitoringのアラートを選択

アラート画面の「Create Policy」からポリシーを作成

Select Metricsから対象のサービスを選択し、作成したログベースの指標を探す
見つからない場合はActiveのチェックを外すと見つかる可能性があります。

ログを検出時に通知する場合は、以下のようにローリングウィンドウ関数を「delta」、
時系列集計を「min」にします。
通知する際にプロジェクトIDやサービス名を通知したい場合は、
時系列のグループ化基準に必要なキーを指定します。(後述のドキュメントで使用します)

一件でも検出すると通知する場合は、以下の設定にします。

チャンネルへの通知設定をします。
Slackを認証・チャンネルを追加していなければ右下の「Manage Notification Channels」から認証・追加してください。

「通知と名前」に移動し、通知したい内容をドキュメントに記載します。
プロジェクトとリソース、サービス名を通知したい場合は画像のように${変数名}で指定します。
(前述した「時系列のグループ化基準」に追加しないと値が空になります。)
Googleドキュメントに変数の指定方法が記載されています。
以上でポリシーを保存します。

間違ったAPIキーでリクエストするなどして通知が届けば完了です。

個人用の小さなアプリだとアクセスを検知できるだけでも、有効な対策にはなると思います。