记录我是如何把rsyslog做成docker镜像，获取nginx的accesslog并且转发到python的
关键点1 nginx日志配置
nginx日志要设置成json格式输出，nginx.conf如下图所示，这个可以在docker镜像中通过volume把nginx.conf挂载进去，然后把/var/log/nginx/access.log挂载到本地
user root;
worker_processes 1;
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
#log_format main '$remote_addr - $remote_user [$time_local] "$request" '
# '$status $body_bytes_sent "$http_referer" '
# '"$http_user_agent" "$http_x_forwarded_for"';
log_format main '{"time_local": "$time_local", '
'"path": "$request_uri", '
'"ip": "$remote_addr", '
'"time": "$time_iso8601", '
'"user_agent": "$http_user_agent", '
'"user_id_got": "$uid_got", '
'"user_id_set": "$uid_set", '
'"remote_user": "$remote_user", '
'"request": "$request", '
'"status": "$status", '
'"body_bytes_sent": "$body_bytes_sent", '
'"request_time": "$request_time", '
'"http_referrer": "$http_referer" }';
access_log /var/log/nginx/access.log main;
sendfile on;
#tcp_nopush on;
keepalive_timeout 65;
#gzip on;
include /etc/nginx/conf.d/*.conf;
}
关键点2 rsyslog配置与Dockerfile
编写一个51-nginx-forward.conf文件放置在/etc/rsyslog.d/下即可
module(load="imfile")
input(type="imfile"
File="/var/log/nginx/access.log"
Tag="mywebsite:")
# omfwd module for forwarding the logs to another tcp server
if( $syslogtag == 'mywebsite:') then {
action(type="omfwd" target="python服务器IP地址" port="6000" protocol="tcp"
action.resumeRetryCount="100"
queue.type="linkedList" queue.size="10000")
}
我们可以用一个Dockerfile来运行rsyslog，docker run的时候注意日志的挂载
FROM ubuntu:16.04
RUN apt-get update && apt-get install -y rsyslog; \
rm -rf /var/lib/apt/lists/*
ADD 51-nginx-forward.conf /etc/rsyslog.d/.
# RUN cat /dev/null> /var/log/mail.log
CMD service rsyslog start && tail -f /var/log/syslog
关键点3 python程序通过tcp的方式读取rsyslog
python程序与rsyslog建立tcp连接，可以实时的进行数据库的插入语句
import asyncio
import json
import time
import database_init
class LogAnalyser:
def __init__(self):
pass
def process(self, str_input):
# print(str_input)
str_input = str_input.decode("utf-8", errors="ignore")
# Add your processing steps here
# ...
try:
# Extract created_at from the log string
str_splits = str_input.split("{", 1)
json_text = "{" + str_splits[1]
data = json.loads(json_text)
created_at = data["time"]
request_all = data["request"].split(" /", 1)
http_type = request_all[0]
path = data["path"]
request_time = data["request_time"]
if PREFIX in data["path"]:
path = data["path"]
return http_type, path, created_at,request_time # The order is relevant for INSERT query params
except Exception as e:
print("error in read_rsylog.py,Class LogAnalyser,function process")
print(e)
return None
@asyncio.coroutine
def handle_echo(reader, writer):
log_filter = LogAnalyser()
while True:
line = yield from reader.readline()
if not line:
break
params = log_filter.process(line)
if params:
# 进行一堆操作，例如进行数据库的插入
# execute_sql(params=params)
if __name__ == '__main__':
CURSOR = database_init.DBConnect().CURSOR
CONN = database_init.DBConnect().CONN
PREFIX = database_init.DBConnect().CONFIG["TEST_SWAGGER"]["PREFIX"]
database_init.DBConnect().create_table()
loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, None, 6000, loop=loop)
server = loop.run_until_complete(coro)
# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
loop.run_forever()
except KeyboardInterrupt:
pass
# Close the server
print("Closing the server.")
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
CURSOR.close()
CONN.close()
