JUNのブログ

JUNのブログ

活動記録や技術メモ

Python Logging SlackHandler の作り方

python で logging を使っていると, たまに 「このログをSlackに飛ばしたいなぁ」ってときがあります.

なのでログをSlackに投稿する Logging Handler 作りました.

Slack側の準備

Incoming Webhook アプリをSlackに導入します.

その他の管理項目 -> Appの管理 でワークスペースのアプリ設定ページにアクセス出来ます.

アプリの設定ページ内のAppディレクトリを検索して Incoming Webhook をSlackに追加します.

f:id:JUN_NETWORKS:20191130005317p:plain
Slack Incoming Webhook

Slackに追加 を押したらどのチャンネルに送信するかというのが出てくるので ログを送信したいチャンネルを選んでください. 選んだらIncoming Webhook インテグレーションの追加 を押してください.

f:id:JUN_NETWORKS:20191130005543p:plain

そうすると Incoming Webhook アプリの設定画面に飛ぶと思います.

その中にある Webhook URL をメモしておいてください.

試しに何かメッセージを送ってみましょう

import requests

webhook_url = "先程メモしたWebHookURL"
requests.post(webhook_url, json={"text": "Pythonから送信したテストメッセージ"})

これを実行すると先程指定したチャンネルにメッセージが来たと思います.

f:id:JUN_NETWORKS:20191130010223p:plain

テキストの投稿以外にも Incoming Webhook が出来ることは沢山ありますが, 今回は扱いません. 気になる人は Slackワークスペース内にある Incoming Webhook アプリ の設定ページを見てください.

SlackHandler を作る

いよいよ本題です. SlackHandler を作っていきます.

SlackHandler を作成する前に Python Loggingモジュールにおける Handler とは何かを理解しないといけません.

Pythonの公式ドキュメント によると

ハンドラは、(ロガーによって生成された) ログ記録を適切な送信先に送ります。

らしいです.

では作っていきましょう.

handlers.py というファイルを作って, 中身を以下の様にします.

import logging

import requests


class SlackHandler(logging.StreamHandler):

    def __init__(self, url):
        super(SlackHandler, self).__init__()
        self.url = url

    def emit(self, record):
        msg = self.format(record)
        self.send_message(msg)

    def send_message(self, text):
        message = {
            'text': text,
        }

        requests.post(self.url, json=message)

コードの解説をします.

SlackHandler が継承している logging.StreamHandler とは[Python公式ドキュメント] から引用すると

logging コアパッケージに含まれる StreamHandler クラスは、ログ出力を sys.stdout, sys.stderr あるいは何らかのファイル風 (file-like) オブジェクト (あるいは、より正確に言えば write() および flush() メソッドをサポートする何らかのオブジェクト) といったストリームに送信します。

つまりは, ログをどこかに送る 基礎的なHandlerです.

def __init__() では スーパークラスである StreamHandlerインスタンスメソッド __init__() を呼び出して StreamHandler の各種変数やメソッドの初期化をします.
また, 引数 url で webhook_url をインスタンス変数に登録しています.

def emit(self, record) は StreamHandler のメソッドで, 例によってPython公式ドキュメントから引用すると,

emit(record)

フォーマッタが指定されていれば、フォーマッタを使ってレコードを書式化します。 次に、レコードが終端記号とともにストリームに書き込まれます。 例外情報が存在する場合、 traceback.print_exception() を使って書式化され、 ストリームの末尾につけられます。

つまりは来たrecordをsetFormatterで設定したformat形式に変換してStreamに記録するというメソッドです. 今回の場合はSlackにログを送信したいのでこのメソッドをオーバーライドすればいいですね.
今回は来たrecordをformatの形式に変形するのはそのままに, その後Slackにメッセージを送信する関数 send_message() を作り, それを使ってformat形式に変形されたログを送信しています.

SlackHandlerを使う

SlackHandler を使う時は他の Logging Handler とほとんど同じように使えます.

以下のコードを実行するとSlackにログが送信されます.

import logging

from handlers import SlackHandler

webhook_url = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

# logger の作成
logger = logging.getLogger(__file__)
logger.setLevel(logging.INFO)
# Slack Handler の作成
slack_handler = SlackHandler(webhook_url)
slack_handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s')
slack_handler.setFormatter(formatter)
logger.addHandler(slack_handler)


def main():
    logger.info("This is Logging Test!!")

    ans = 1 + 2
    logger.info(f"1 + 2 = {ans}")


if __name__ == "__main__":
    main()

f:id:JUN_NETWORKS:20191130031947p:plain
Slackに送られてきたメッセージ

SlackHandler (応用編)

SlackHandlerの改造

先程作成したSlackHandlerは最低限の機能を満たしていますが, 私が実際に使っているものはさらに改造を施したものです. 折角なのでご紹介します.

class SlackHandler(logging.StreamHandler):

    def __init__(self, url, username="", icon_emoji=":computer:"):
        super(SlackHandler, self).__init__()
        self.url = url
        self.username = username
        self.icon_emoji = icon_emoji

    def emit(self, record):
        msg = self.format(record)
        self.send_message(msg, record.levelno)

    def send_message(self, text, level_num):
        icon_emoji = self.icon_emoji
        if level_num > 30:  # Warning 以上
            # メンションを飛ばす
            text = "<@MEMBERID> " + text
            icon_emoji = ":exclamation:"

        message = {
            'text': text,
            'username': self.username,
            'icon_emoji': icon_emoji
        }

        requests.post(self.url, json=message)

emojiusername に関しては Slackのワークスペース内の Incoming Webhookアプリの説明に書いてあるのでそっちを見てほしいです. また, SlackのIncoming Webhooksを使い倒す - Qiita もわかりやすくていいと思います.

それではそれ以外の部分について解説します.

send_message(self, text, levelno) でlogging のレベルを渡しています. これにより, levelごとに処理を分けることが出来ます.
今回の場合は, WARNING以上のログはWebHookアプリのアイコンを❗に変更し, 私のアカウントに対してメンションを付けてログをSlackに送信しています. この部分をChannelIDに変えるとチャンネルに参加している全員に通知を飛ばせます.

f:id:JUN_NETWORKS:20191130164909p:plain
WARNING 以上のレベルは アイコンが変わる & ユーザーにメンションされる

メンションを飛ばす部分については SlackのIncoming Webhooksでメンションを飛ばす方法 - Qiita を見てください.

SlackHandler をもう少し高度に使う

先程の仕様例でもいいのですが, あれは少しスマートじゃないですし, 実際に使う際はHandlerやFormatを dictConfigfileConfig で管理すると思います. 自分の場合は dictConfig をいつも使っているので, 今回はそれで説明します.

私は dictConfig を使って. 以下のように設定しています.

import logging
import logging.config


# logging
LOGGING = {
    'version': 1,
    'disable_existing_loggers': True,
    'formatters': {
        'standard': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        },
    },
    'handlers': {
        'default': {
            'level': 'INFO',
            'formatter': 'standard',
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',  # Default is stderr
        },
        # ファイル出力ハンドラー
        'file': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': "test.log",
            'formatter': 'standard',
        },
        'slack': {
            'level': 'INFO',
            'class': 'handlers.SlackHandler',
            'url': os.environ.get("SLACK_WEBHOOK_URL"),
            'username': 'Landing Classification Observer',
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['default', 'file', 'slack'],
            'level': 'INFO',
            'propagate': False
        },
    }
}


logging.config.dictConfig(LOGGING)
logger = logging.getLogger(__name__)

全ての設定を LOGGING で管理出来るようになりました.

ここでのポイントは

  • SlackのWebHook URLを環境変数で管理している.
    WebHook URLなどはバージョン管理などに入れたくない情報だと思うので, 環境変数に入れています. こちらのほうがセキュリティ的にもいいと思います.

  • SlackHandler のインスタンス作成時に必要な情報(引数) を全て LOGGING 内で完結している.
    コードの見た目的にも綺麗です.

おまけ

ここからは SlackHandler に必要な訳ではないが, あると便利かも? しれないものを紹介します.

SlackHandler用に新しいレベルを作る

標準出力ハンドラー と ファイルハンドラー ではログを出力したいけど, SlackHandlerには出力しなくない,,,
だからといって SlackHandler のレベルを WARNING に設定したくはない...

ってなった時には 新しいレベルを作りましょう!!

ちなみに python の logging の標準で用意されているレベルは以下のようになっています. Python 公式ドキュメントより以下の様になっています

Level Numeric value
CRITICAL 50
ERROR 40
WARNING 30
INFO 20
DEBUG 10
NOTSET 0

今回は INFO ってほど気軽な感じじゃないけど, WARNINGを使うほどでもない... けど Slackに送信してほしいので INFO と WARNING の間を取って レベル25 の IMPORTANT というレベルを作りましょう.

# add new level called IMPORTANT
IMPORTANT_LEVEL_NUM = 25
logging.addLevelName(IMPORTANT_LEVEL_NUM, "IMPORTANT")


def important(self, message, *args, **kws):
    if self.isEnabledFor(IMPORTANT_LEVEL_NUM):
        # Yes, logger takes its '*args' as 'args'.
        self._log(IMPORTANT_LEVEL_NUM, message, args, **kws)


logging.Logger.important = important

これで IMPORTANT という新しいレベルが logging に登録されました.

使う際には以下の様に Handler などを設定する前に定義してから使います.

import logging
import logging.config


# logging settings
# add new level called IMPORTANT
IMPORTANT_LEVEL_NUM = 25
logging.addLevelName(IMPORTANT_LEVEL_NUM, "IMPORTANT")


def important(self, message, *args, **kws):
    if self.isEnabledFor(IMPORTANT_LEVEL_NUM):
        # Yes, logger takes its '*args' as 'args'.
        self._log(IMPORTANT_LEVEL_NUM, message, args, **kws)


logging.Logger.important = important

# Define logging dict
# logging
LOGGING = {
    'version': 1,
    'disable_existing_loggers': True,
    'formatters': {
        'standard': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        },
    },
    'handlers': {
        'default': {
            'level': 'INFO',
            'formatter': 'standard',
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',  # Default is stderr
        },
        # ファイル出力ハンドラー
        'file': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': "test.log",
            'formatter': 'standard',
        },
        'slack': {
            'level': 'IMPORTANT',
            'class': 'handlers.SlackHandler',
            'url': os.environ.get("SLACK_WEBHOOK_URL"),
            'username': 'Landing Classification Observer',
            'formatter': 'standard',
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['default', 'file', 'slack'],
            'level': 'INFO',
            'propagate': False
        },
    }
}


logging.config.dictConfig(LOGGING)
logger = logging.getLogger(__name__)


def main():
    logger.info("This is Logging Test!!")

    ans = 1 + 2
    logger.important(f"1 + 2 = {ans}")


if __name__ == "__main__":
    main()

これを実行すると, 標準出力とファイル出力ハンドラーに対しては 2つのログが出力されます.

f:id:JUN_NETWORKS:20191130155400p:plain
標準出力

出力された test.log の中身は以下のようになっています.

2019-11-30 15:53:35,869 [INFO] __main__: This is Logging Test!!
2019-11-30 15:53:35,870 [IMPORTANT] __main__: 1 + 2 = 3

そして, 肝心のSlackの方ですが, 以下のように IMPORTANT のログのみが送信されています.

f:id:JUN_NETWORKS:20191130155605p:plain
Slackに送られてきたメッセージ

いい感じですね!!

画像をSlackに自動投稿する

これは色んな所で既出なのであんまりガッツリは書きません. 詳しい解説は他のサイトを見てください.

コードだけ貼ります.

import requests


def upload_image(token, image, filename="", channels="logging", title="", initial_comment=''):
    """ 画像をSlackワークスペース内のチャンネルにアップロードする.
    別に画像以外でもいける

    Parameters
    ----------
    token : str
        Slack API Token.
        取得方法はggって
    image : binary
        画像のバイナリーデータ.
        ファイルから読み込む時は  open("filename", "rb")
        np.ndarray形式で変数imgに保持している時は  cv2.imencode('.jpg', img)[1].tostring()
    filename : str, optional
        slackに投稿される画像のファイル名, by default ""
    channels : str, optional
        投稿先のチャンネル名, by default "logging"
    title : str, optional
        Slack上でのタイトル, by default ""
    initial_comment : str, optional
        コメント(メッセージ), by default ''
    """
    files = {'file': image}
    params = {
        'token': token,
        'channels': channels,
        'filename': filename,
        'title': title,
        'initial_comment': initial_comment
    }
    url = "https://slack.com/api/files.upload"
    requests.post(url=url, params=params, files=files)

上記のコードでわからなかった場合は以下の記事読んでください.

qiita.com

あとがき

今回は Slack にログを送信できる SlackHandler を作成しました.

そもそもこれを作ろうと思ったのは機械学習の学習の学習の経過をSlackに送信したかったからです. そこで方法を調べたのですが, 多くの記事が別に関数を作って実装していました. 個人的には ログ関係は全て Logging にまとめてしまいたかったので今回SlackHandlerを作成しました. 調べた感じSlackHandlerについて書いてある記事は 日本語,英語 共に無かったのでもしかして初かな? 初だと少し嬉しいかもですね.

また, おまけの 新しいレベルを作る 部分は正直必須ではないです. 昔結構苦労して作ったのに現在使っていなくて悲しいので作り方を忘れる前に残しておきたかったので書きました. またどこかで使うかもしれないのでね.

もし, 参考になったり, 使って便利だったらこの記事のリンクを貼ったり, 共有してくれたら嬉しいです. 結構書くの時間かかったので.

てか今思ったんですけど, Slack APP に WebHook と uploadfile の権限持たせれば1つに統合出来るやん.