はじめに

こちらは、「富士通クラウドテクノロジーズ Advent Calendar 2023」の24日目の記事です。
昨日の記事は@hasunuma さんの、「FJCTF#2(社内CTF)を開催しました！」でした。

本日のテーマ

私は普段、Slackでサーバーのアラート通知メールを目にする機会があるのですが、見慣れないアラートの調査に時間がかかっていました。
そんな中で、過去の調査記録がいつでも参照できると原因の調査が楽になるだろうと思いました。
そこで、アラート通知メールが届いたら過去の調査記録を検索しSlack上に投稿する仕組みを作ることをテーマにしました。

前提

過去の調査記録はGitlabのIssueで管理しているため、投稿されるのはGitlabのIssueのURLとします。
アラート通知メールをSlackに送る仕組みは既にあったのでスキップします。

どのような流れで記録を投稿するか

まず大まかに以下のSTEPで調査記録を投稿することを決めました。

Bolt for Python でSlack上のメール投稿をリッスン
メールの情報から、特定のアラートパターンに一致しているかチェック
一致している場合、今度はアラートパターンに一致するタイトルのIssueを検索
マッチしたIssueのURLをSlackのアラートメール投稿に返信

ここから、各STEPの実装に入ります。

1. Bolt for Python でSlack上のメール投稿をリッスン

メッセージをリッスンする

まずメッセージイベントを取得できるように、チュートリアルのページを参考にコードを書いてみます。

run.py

import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import json

app = App(token=os.environ.get("SLACK_BOT_TOKEN"))

# メッセージイベントをリッスン
@app.event("message")
def listen_message(message, client):
    print(json.dumps(message,indent=4))

# アプリを起動します
if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

メッセージをSlackで投稿してみます。

無事にメッセージが取得できていることがわかりました。

python3 run.py 
⚡️ Bolt app is running!
{
    "client_msg_id": "f87beb9a-d2de-4e2e-aa80-f9caf56c010f",
    "type": "message",
    "text": "test",
    "user": "xxx",
    "ts": "1703431443.741819",
    "blocks": [
        {
            "type": "rich_text",
            "block_id": "gB9fq",
            "elements": [
                {
                    "type": "rich_text_section",
                    "elements": [
                        {
                            "type": "text",
                            "text": "test"
                        }
                    ]
                }
            ]
        }
    ],
    "team": "xxx",
    "channel": "xxx",
    "event_ts": "1703431443.741819",
    "channel_type": "channel"
}

メール投稿をリッスンする

実際にはアラート通知メールを読み取りたいので、メッセージのsubtypeがfile_shareの場合に条件に入るようにします。
※メール投稿はファイル共有扱いのため

run.py

# メッセージイベントをリッスン
@app.event("message")
def listen_message(message, client):

  # messageのsubtypeがファイル共有だった場合
  if message.get('subtype') == 'file_share':
    print(json.dumps(message,indent=4))

メールを送信します。

送信されたメールの情報が取得できました。

python3 run.py 
⚡️ Bolt app is running!
{
    "type": "message",
    "text": "",
    "files": [
        {
            "id": "xxx",
            "created": 1703440595,
            "timestamp": 1703440586,
            "name": "test",

...

            "permalink_public": "https://slack-files.com/xxx-xxx-xxx",
            "subject": "test",
            "to": [

...

            "file_access": "visible"
        }
    ],
    "upload": true,
    "user": "USLACKBOT",
    "display_as_bot": true,
    "bot_id": "xxx",
    "ts": "1703440595.714489",
    "channel": "xxx",
    "subtype": "file_share",
    "event_ts": "1703440595.714489",
    "channel_type": "channel"
}

この中で、アラートタイトルの取得にsubject、スレッドに返信するためにevent_tsを使用します。

2. メールの情報から、特定のアラートパターンに一致しているかチェック

いくつかのアラートのパターンを用意し、メールの件名とマッチした場合に処理を行うようにしてみます。

run.py

import re

# 正規表現をコンパイル
patterns = [re.compile(r'.* PROBLEM: Zabbix agent on .* is unreachable for 5 minutes'),re.compile(r'PROBLEM: Too many deadlock DB error on .+'),r'PROBLEM: Error in API log \(.+\) on .+']

# メッセージイベントをリッスン
@app.event("message")
def listen_message(message, client):

  # messageのsubtypeがファイル共有だった場合
  if message.get('subtype') == 'file_share':

    # メールの件名を取得
    subject = message['files'][0]['subject']

    for pattern in patterns:

      # 特定のパターンとマッチした場合
      if(re.search(pattern, subject)):

        # マッチした件名を表示
        print(subject)

        # マッチしたパターンを表示
        print(pattern)

メールを送ってみます。

無事にアラートのパターンとメールの件名がマッチしたときに処理を行うようにできました。

python3 run.py 
⚡️ Bolt app is running!
PROBLEM: Too many deadlock DB error on hoge
re.compile('PROBLEM: Too many deadlock DB error on .+')
hoge PROBLEM: Zabbix agent on hoge is unreachable for 5 minutes
re.compile('.* PROBLEM: Zabbix agent on .* is unreachable for 5 minutes')

3. 一致している場合、今度はアラートパターンに一致するタイトルのIssueを検索

GitlabのAPIでは、正規表現を用いて特定のタイトルのIssueを検索する機能はサポートされていません。
そのため、一度全てのIssueを取得した後に、正規表現でアラートのパターンと一致するタイトルのみに絞り込みます。

Gitlabの対象プロジェクトの全てのIssueを取得する

run.py

import urllib
import requests

# GitlabのAPI使用に必要な定数を定義
GITLAB_URL = "https://gitlab.hoge"
PROJECT_PATH = "fuga/piyo"
project_id = urllib.parse.quote(PROJECT_PATH, safe='')
private_token = os.environ['gitlab_private_token']

# メッセージイベントをリッスン
@app.event("message")
def listen_message(message, client):

  # messageのsubtypeがファイル共有だった場合
  if message.get('subtype') == 'file_share':

    # メールの件名を取得
    subject = message['files'][0]['subject']

    for pattern in patterns:

      # 特定のパターンとマッチした場合
      if(re.search(pattern, subject)):

        # GitlabのAPI使用に必要な変数を定義
        url = f'{GITLAB_URL}/api/v4/projects/{project_id}/issues'
        headers = {'PRIVATE-TOKEN': private_token}
        params = {"per_page": 100}
        all_issues = []
        page = 1

        # 全てのIssueを取得できるまでループ
        while True:
            params["page"] = page
            response = requests.get(url, headers=headers, params=params)
            issues = response.json()
            if not issues:
                break
            all_issues.extend(issues)
            page += 1

    print(len(all_issues))

無事にIssuesの要素数925が取得できました。

python3 run.py 
⚡️ Bolt app is running!
925

アラートパターンに一致するタイトルのIssueに絞り込む

先ほど取得した全てのIssueから、正規表現のパターンにマッチするIssueのみに絞り込みます。

run.py

        # 正規表現にマッチするIssueを検索
        matched_issues = [issue for issue in all_issues if re.search(pattern, issue["title"])]

        for matched_issue in matched_issues:
          print(json.dumps(matched_issue))

無事にマッチしたIssueのみの情報が取得できました。

python3 run.py 
⚡️ Bolt app is running!
{
    "id": 168453,
    "iid": 768,
    "project_id": 3203,
    "title": "PROBLEM: Error in API log (xxx) on xxx",

...

    "issue_type": "issue",
    "web_url": "https://gitlab.hoge/fuga/piyo/-/issues/768",
    "time_stats": {

...

    "iteration": null
}

4. マッチしたIssueのURLをSlackのアラートメール投稿に返信

ここまで来たら、あとはIssueのURLをテキストにくっつけて投稿するだけですね。

run.py

        # メッセージを投稿するチャンネルのID
        channel_id = "xxx"

        # 正規表現にマッチするIssueを検索
        matched_issues = [issue for issue in all_issues if re.search(pattern, issue["title"])]

        text = f"`{pattern}` のパターンのIssueが `{len(matched_issues)}` 件見つかったよ！"

        # マッチしたIssueのURLを返信用テキストに追加
        for matched_issue in matched_issues:
          text += '\n' + matched_issue['web_url']

        client.chat_postMessage(
            channel=channel_id,
            thread_ts=message['event_ts'],
            text=text
        )

ダミーのアラートメールを送ったところ、無事にアラートの条件にマッチするIssueが投稿されました。

最後に

蓄積されたノウハウに気軽にアクセスできる状態になったので、今後の作業効率アップに繋がると嬉しいです。
今回はメール通知のタイトルのみ使用しましたが、本文まで含めて処理の条件を決めると色々なことに応用できそうですね。

明日は @ConHumi さんがETFレポートについて書かれるそうです。
お楽しみに！