Python日志监控系统是后端服务运维中非常重要的组成部分，它能够帮助我们实时掌握服务的运行状态，快速定位线上问题。一个完整的日志监控系统通常包含日志采集、日志传输、日志处理、日志存储和告警展示几个核心模块，每个模块的实现逻辑都和Python的相关生态库紧密相关。
日志监控系统核心原理
1. 日志采集原理
日志采集是监控系统的第一步，Python中常用的日志采集方式有两种，一种是通过logging模块主动输出日志，另一种是通过读取服务已经生成的日志文件获取内容。主动输出日志的方式可以在代码层面控制日志的格式和级别，方便后续处理；读取日志文件的方式则适合对接已经运行的老服务，不需要修改原有代码。
2. 日志传输与处理原理
采集到的日志通常需要经过传输和处理环节，传输可以使用消息队列或者直接的网络请求，处理环节主要完成日志的过滤、格式化和结构化。比如我们可以过滤掉级别为DEBUG的无用日志，把非结构化的文本日志转换成包含时间、服务名、日志级别、日志内容的结构化数据，方便后续存储和查询。
3. 存储与告警原理
处理后的结构化日志一般会存储到Elasticsearch这类支持全文检索的数据库中，方便后续快速查询。告警模块则会定时检查存储的日志数据，当发现符合告警规则的日志（比如错误日志数量超过阈值）时，通过邮件、企微消息等方式通知运维人员。
实战案例：搭建简易日志监控系统
环境准备
本案例需要安装的Python库如下：
- logging：Python内置日志库，用于输出日志
- watchdog：用于监控日志文件的变化
- requests：用于发送告警请求
步骤1：实现日志采集模块
我们使用watchdog监控指定目录下的日志文件，当文件有新内容写入时，自动读取新增的日志行。代码如下：
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class LogHandler(FileSystemEventHandler):
def __init__(self, log_path):
self.log_path = log_path
self.last_position = 0
def on_modified(self, event):
# 只处理日志文件的修改事件
if event.src_path == self.log_path:
with open(self.log_path, 'r', encoding='utf-8') as f:
f.seek(self.last_position)
new_lines = f.readlines()
self.last_position = f.tell()
for line in new_lines:
# 将采集到的日志传递给处理模块
process_log(line.strip())
def start_collect(log_path):
event_handler = LogHandler(log_path)
observer = Observer()
# 监控日志文件所在的目录
observer.schedule(event_handler, path=log_path.rsplit('/', 1)[0], recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
步骤2：实现日志处理模块
处理模块负责过滤和结构化日志，这里假设我们的日志格式为[时间] [级别] 服务名 - 日志内容，处理代码如下：
import re
def process_log(raw_log):
# 过滤空日志
if not raw_log:
return
# 匹配日志格式的正则表达式
pattern = r'[(.*?)] [(.*?)] (.*?) - (.*)'
match = re.match(pattern, raw_log)
if match:
log_time, level, service_name, content = match.groups()
# 只处理ERROR级别的日志
if level == 'ERROR':
structured_log = {
'time': log_time,
'level': level,
'service_name': service_name,
'content': content
}
# 将结构化日志传递给存储和告警模块
save_log(structured_log)
check_alarm(structured_log)
else:
print(f'无法解析的日志: {raw_log}')
步骤3：实现日志存储与告警模块
这里为了简化，存储模块我们暂时把日志打印到控制台，告警模块模拟发送请求到告警接口，实际项目中可以替换为Elasticsearch存储和真实的告警接口。代码如下：
import requests
# 模拟存储日志
def save_log(log_data):
print(f'存储日志: {log_data}')
# 模拟告警检查，这里简单判断错误日志就触发告警
def check_alarm(log_data):
alarm_url = 'http://ipipp.com/alarm'
alarm_data = {
'title': f'{log_data["service_name"]}服务出现错误日志',
'content': f'时间: {log_data["time"]}, 内容: {log_data["content"]}'
}
try:
response = requests.post(alarm_url, json=alarm_data, timeout=5)
if response.status_code == 200:
print('告警发送成功')
else:
print(f'告警发送失败，状态码: {response.status_code}')
except Exception as e:
print(f'告警发送异常: {e}')
步骤4：运行测试
我们首先启动日志采集程序，然后向监控的日志文件中写入一条ERROR级别的日志，观察是否触发存储和告警逻辑。测试代码如下：
if __name__ == '__main__':
# 指定要监控的日志文件路径
test_log_path = './test_service.log'
# 先写入一条测试日志
with open(test_log_path, 'a', encoding='utf-8') as f:
f.write('[2024-05-20 10:00:00] [ERROR] user_service - 用户登录接口调用失败n')
# 启动日志采集
start_collect(test_log_path)
运行上述代码后，控制台会输出存储的日志信息，同时模拟发送告警请求，说明整个监控系统已经正常运行。实际项目中可以根据需求扩展功能，比如增加日志级别过滤规则、对接真实的存储和告警组件、增加日志查询页面等。