watchdogで即検知！NGINXアクセスログをPythonでリアルタイム監視してInfluxDBに記録する

pip install watchdog

# =========================================
# 共通ログフォーマット定義（http ブロック内に記述）
# =========================================
log_format detailed '$remote_addr - $remote_user [$time_local] "$request" '
                    '$status $body_bytes_sent "$http_referer" '
                    '"$http_user_agent" "$http_x_forwarded_for"';

server {
    listen 443 ssl;
    server_name <yourdomain>;

    # ...（SSL設定など）...

    # detailed フォーマットでログを出力
    access_log  /srv/nginx/log/access.log detailed;
    error_log   /srv/nginx/log/error.log;

    # ...
}

sudo nginx -t          # 設定ファイルの文法チェック
sudo systemctl reload nginx

# -*- coding: utf-8 -*-
import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from influx.v2 import writer as influxwriter
import smarthomelog

# Influxライターとログ出力インスタンスを初期化
influx = influxwriter.Writer()
success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)

# ===== ログファイルのパス設定 =====
LOG_DIR = '/srv/nginx/log'
ACCESS_LOG = os.path.join(LOG_DIR, 'access.log')
ERROR_LOG  = os.path.join(LOG_DIR, 'error.log')

# ===== 各ファイルの読み取り位置を管理する辞書 =====
file_positions = {}

def read_new_lines(file_path, log_type):
    """ファイルの末尾から新しい行だけ読む"""
    pos = file_positions.get(file_path, 0)
    with open(file_path, 'r') as f:
        f.seek(pos)
        new_lines = f.readlines()
        file_positions[file_path] = f.tell()  # 読み取り位置を更新

    for line in new_lines:
        if line.strip():
            influx.insert_nginx_log(log_type, line)

    success_log.write('処理が正常に終了しました。', __file__)

class LogFileHandler(FileSystemEventHandler):
    """ログファイル変更イベントハンドラ"""
    def on_modified(self, event):
        if event.is_directory:
            return
        if event.src_path == ACCESS_LOG:
            read_new_lines(ACCESS_LOG, 'ACCESS')
        elif event.src_path == ERROR_LOG:
            read_new_lines(ERROR_LOG, 'ERROR')

if __name__ == "__main__":
    # 起動時に既存のログ末尾位置を記録（既存行はスキップ）
    for log_file in [ACCESS_LOG, ERROR_LOG]:
        if os.path.exists(log_file):
            with open(log_file, 'r') as f:
                f.seek(0, os.SEEK_END)
                file_positions[log_file] = f.tell()

    # watchdog 監視を開始
    event_handler = LogFileHandler()
    observer = Observer()
    observer.schedule(event_handler, LOG_DIR, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

def insert_nginx_log(self, log_type: str, message: str):
    measurement = 'nginx_log'
    tags = {
        'log_type': log_type   # 'ACCESS' または 'ERROR'
    }
    fields = {
        'message': message,
        'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
    }
    try:
        self.conn.write_dict(measurement, tags, fields)
    except Exception as e:
        print(e)

for log_file in [ACCESS_LOG, ERROR_LOG]:
    if os.path.exists(log_file):
        with open(log_file, 'r') as f:
            f.seek(0, os.SEEK_END)
            file_positions[log_file] = f.tell()

def read_new_lines(file_path, log_type):
    pos = file_positions.get(file_path, 0)
    with open(file_path, 'r') as f:
        f.seek(pos)           # 前回読み終えた位置から再開
        new_lines = f.readlines()
        file_positions[file_path] = f.tell()  # 読み取り位置を更新

class LogFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        if event.src_path == ACCESS_LOG:
            read_new_lines(ACCESS_LOG, 'ACCESS')
        elif event.src_path == ERROR_LOG:
            read_new_lines(ERROR_LOG, 'ERROR')

[Unit]
Description=NGINX Log Record to InfluxDB
After=network.target influxd.service nginx.service

[Service]
ExecStart=/usr/bin/python3 /home/<username>/projects/smarthome/core/record/raspi/nginx_log_record.py
Restart=always
RestartSec=10
User=<username>

[Install]
WantedBy=multi-user.target

sudo cp nginx_log_record.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable nginx_log_record
sudo systemctl start nginx_log_record

python3 nginx_log_record.py

from(bucket: "your-bucket")
  |> range(start: -5m)
  |> filter(fn: (r) => r._measurement == "nginx_log")
  |> filter(fn: (r) => r._field == "message")

sudo systemctl status nginx_log_record