監視・通知

DjangoのログをSlackにリアルタイム送信する — カスタムLogging Handler実装

Django Slack ログ監視

DjangoのログをSlackにリアルタイム送信する
カスタムLogging Handler実装

攻撃を受けたとき、エラーが発生したとき、すぐに気づけていますか?
Djangoの標準loggingモジュールにカスタムハンドラを追加するだけで、ログをSlackにリアルタイム送信できます。
追加パッケージ不要、Python標準ライブラリだけで実装します。

こんな人向けの記事です

  • Djangoアプリのエラーや攻撃をすぐに知りたい
  • ログの種類ごとに別のSlackチャンネルに送りたい
  • 追加パッケージなしで実装したい
  • Pythonのloggingモジュールの仕組みを理解したい

アーキテクチャ

Djangoの標準loggingは、ロガー → ハンドラ → フォーマッタの3層構造です。ここにSlack送信ハンドラを追加します。

ログ発生
security / error / access
既存ハンドラ
ファイル出力
+
Slackハンドラ
chat.postMessage
Slackチャンネル
#homepage-security 等

既存のファイル出力はそのままに、Slackハンドラを並列に追加するだけです。

カスタムハンドラの実装

logging.Handlerを継承し、Slack APIのchat.postMessageを呼び出すハンドラを作成します。

config/slack_log_handler.py
import json
import logging
import threading
import urllib.request
import urllib.error
import os


class SlackLogHandler(logging.Handler):
    """Slack chat.postMessage API を使ってログを送信するハンドラ"""

    COLORS = {
        logging.CRITICAL: '#dc3545',
        logging.ERROR: '#dc3545',
        logging.WARNING: '#ffc107',
        logging.INFO: '#28a745',
        logging.DEBUG: '#6c757d',
    }

    def __init__(self, channel, token=None, **kwargs):
        super().__init__(**kwargs)
        self.channel = channel
        self.token = token or os.environ.get('SLACK_BOT_TOKEN', '')

    def emit(self, record):
        if not self.token:
            return
        try:
            msg = self.format(record)
            thread = threading.Thread(
                target=self._send, args=(msg, record.levelno), daemon=True
            )
            thread.start()
        except Exception:
            self.handleError(record)

    def _send(self, message, levelno):
        color = self.COLORS.get(levelno, '#6c757d')
        payload = json.dumps({
            'channel': self.channel,
            'attachments': [{
                'color': color,
                'text': message,
                'fallback': message,
            }],
        }).encode('utf-8')

        req = urllib.request.Request(
            'https://slack.com/api/chat.postMessage',
            data=payload,
            headers={
                'Content-Type': 'application/json; charset=utf-8',
                'Authorization': f'Bearer {self.token}',
            },
        )

        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                body = json.loads(resp.read().decode('utf-8'))
                if not body.get('ok'):
                    print(f'[SlackLogHandler] API error: {body.get("error")}')
        except (urllib.error.URLError, OSError) as e:
            print(f'[SlackLogHandler] Send failed: {e}')

設計のポイント

設計理由
threading.Threadで非同期送信Slack APIの応答を待たずにリクエスト処理を継続
daemon=Trueメインプロセス終了時にスレッドも終了
urllib.requestを使用追加パッケージ不要(Python標準ライブラリ)
エラーはprintで出力ログハンドラ内でログを出すと無限ループするため
ログレベルに応じた色分けSlackで視覚的に重要度を判別できる
無限ループに注意

Slack送信の失敗をDjangoのloggerで記録すると、その記録が再度Slack送信を呼び、無限ループになります。
失敗時はprint()でコンソール出力に留めてください。

settings.pyへの組み込み

環境変数SLACK_BOT_TOKENが設定されている場合のみハンドラを有効化します。

config/settings.py
# Slack通知設定
SLACK_BOT_TOKEN = os.environ.get('SLACK_BOT_TOKEN', '')

# SLACK_BOT_TOKEN が設定されている場合のみ有効化
if SLACK_BOT_TOKEN:
    LOGGING['handlers'].update({
        'slack_security': {
            'level': 'WARNING',
            'class': 'config.slack_log_handler.SlackLogHandler',
            'channel': '#homepage-security',
            'token': SLACK_BOT_TOKEN,
            'formatter': 'verbose',
        },
        'slack_error': {
            'level': 'ERROR',
            'class': 'config.slack_log_handler.SlackLogHandler',
            'channel': '#homepage-error',
            'token': SLACK_BOT_TOKEN,
            'formatter': 'verbose',
        },
    })
    LOGGING['loggers']['security']['handlers'].append('slack_security')
    LOGGING['loggers']['django.request']['handlers'].append('slack_error')
トークン未設定時は自動無効化

SLACK_BOT_TOKENが空の場合、Slackハンドラは追加されず、既存のファイル出力のみで動作します。
開発環境でSlack通知が不要な場合は環境変数を設定しないだけでOKです。

チャンネル設計

ログの種類ごとにチャンネルを分けることで、通知の見逃しを防ぎます。

チャンネルログ種別レベル内容
#homepage-securitysecurityWARNINGXSS, SQLi, Traversal等の攻撃検知
#homepage-errordjango.requestERRORHTTP 5xxエラー
#homepage-performanceperformanceWARNING2秒超の遅いリクエスト
#homepage-accessaccessINFO全HTTPアクセスログ
#homepage-generaldjangoWARNING一般的なDjango警告

Docker環境での設定

.env
SLACK_BOT_TOKEN=xoxb-xxxxxxxxxxxx-xxxxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx
docker-compose.yml
services:
  web:
    env_file:
      - .env
    volumes:
      - ./log:/app/log    # ログをホスト側にもマウント
ログディレクトリの権限

Dockerでログディレクトリをマウントする場合、コンテナ内のユーザーに書き込み権限が必要です。

chmod 777 ./log
chmod 666 ./log/*.log

動作確認

Django shellからテストログを送信して、Slackに届くか確認します。

Django Shell
import logging
security_logger = logging.getLogger('security')
security_logger.warning('テスト: %s,%s,%s,%s,%s,%s',
    'XSS Attack', '192.168.1.1', 'test', 'GET', '/', '<script>'
)

#homepage-securityチャンネルに黄色(WARNING)のメッセージが届けば成功です。

まとめ

  • logging.Handlerを継承したカスタムハンドラでSlack送信を実装
  • threadingで非同期送信し、リクエスト処理をブロックしない
  • SLACK_BOT_TOKEN環境変数でトークンを管理(未設定時は自動無効化)
  • ログ種別ごとにSlackチャンネルを分けて通知の見逃しを防止
  • 追加パッケージ不要、Python標準ライブラリ(urllib)のみで実装