实时获取任务请求对应的Nginx日志

要在Python中实时获取特定任务请求对应的Nginx日志，有几种可行的方法。下面我将介绍几种常见方案：

方法一：使用Nginx日志模块和文件监控

import os
import time
import re
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class NginxLogHandler(FileSystemEventHandler):
    def __init__(self, log_file, request_id_pattern):
        self.log_file = log_file
        self.request_id_pattern = request_id_pattern
        self.current_position = os.path.getsize(log_file)

    def on_modified(self, event):
        if event.src_path == self.log_file:
            with open(self.log_file, 'r') as f:
                f.seek(self.current_position)
                new_lines = f.readlines()
                self.current_position = f.tell()

                for line in new_lines:
                    if re.search(self.request_id_pattern, line):
                        print(f"Found matching log: {line.strip()}")

# 配置参数
log_file = '/var/log/nginx/access.log'
request_id = 'your-task-request-id'  # 替换为你的请求ID或标识
request_id_pattern = re.compile(request_id)

# 创建观察者
event_handler = NginxLogHandler(log_file, request_id_pattern)
observer = Observer()
observer.schedule(event_handler, path=os.path.dirname(log_file))
observer.start()

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

方法二：使用Nginx的syslog和Python日志接收

1. 首先配置Nginx发送日志到syslog：

error_log syslog:server=127.0.0.1:514 debug;
access_log syslog:server=127.0.0.1:514,tag=nginx_access main;

2. Python代码接收syslog：

import socketserver
import re

class SyslogUDPHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = bytes.decode(self.request[0].strip())
        if 'nginx_access' in data and 'your-task-request-id' in data:
            print(f"Matched log: {data}")

if __name__ == "__main__":
    HOST, PORT = "0.0.0.0", 514
    with socketserver.UDPServer((HOST, PORT), SyslogUDPHandler) as server:
        server.serve_forever()

方法三：使用Nginx Lua模块实时处理

如果你能修改Nginx配置，可以使用Lua模块实时处理日志：

http {
    lua_shared_dict task_logs 10m;

    server {
        location /your-task-endpoint {
            access_by_lua_block {
                ngx.shared.task_logs:set(ngx.var.request_id, ngx.var.time_iso8601)
            }

            # 你的正常处理逻辑
        }

        location /get-task-log {
            content_by_lua_block {
                local request_id = ngx.var.arg_request_id
                local log_time = ngx.shared.task_logs:get(request_id)
                ngx.say("Task ", request_id, " was accessed at ", log_time)
            }
        }
    }
}

方法四：使用Fluentd/Prometheus等日志收集系统

对于生产环境，建议使用专业的日志收集系统：

1. 配置Nginx输出日志到Fluentd
2. 使用Python订阅Fluentd的输出

from fluent import sender
from fluent import event

# 设置Fluentd连接
logger = sender.FluentSender('nginx', host='localhost', port=24224)

# 发送自定义事件
event.Event('task.request', {
    'request_id': 'your-task-request-id',
    'message': 'Task started'
})

# 接收端可以使用Fluentd的forward或http插件获取数据

最佳实践建议

1. 在Nginx配置中添加请求ID：

log_format custom '$remote_addr - $remote_user [$time_local] '
                  '"$request" $status $body_bytes_sent '
                  '"$http_referer" "$http_user_agent" '
                  '"$http_x_request_id"';
access_log /var/log/nginx/access.log custom;

2. 在应用层生成唯一请求ID并传递到Nginx：

import requests
import uuid

request_id = str(uuid.uuid4())
headers = {'X-Request-ID': request_id}
response = requests.get('http://your-service.com', headers=headers)

选择哪种方法取决于你的具体需求、系统架构和权限级别。对于简单场景，方法一就足够了；对于复杂生产环境，建议使用方法四。