背景：已知某后台服务将日志存放在本地硬盘的日志文件中，该服务也支持代码热更新，并在完成热更新后输出一条日志。我们需要对服务日志进行监控，以确保文件热更新后的错误能被第一时间发现。
我们提供 Python 程序模拟（https://pastebin.com/pZH8wruC，密码pWXRFeSpwU）上述行为，该程序会不断生成日志，并输出到日志文件 1.log 中，日志格式参考源码。
要求：
1、使用 Python 或者 Shell 对上述日志进行检测，检测方式可以是常驻进程实时检测，也可以通过系统定时任务（需要说明如何配置）定时检测日志；
2、当检测到热更新日志时（以"load file:"开头，具体格式参考源码），对错误日志（error级别）进行分类统计（分类按照日志内容进行）；
3、如果热更新后1分钟内某一类别的错误日志比热更新前1分钟内的量级更多（增加50%），则在另一个文件 2.log 中记录相关日志（附带该量级超标的日志内容、热更新前后的量级信息）。
//这是链接里面的代码
# 日志生成用例
# - 程序随机地生成 info 和 error 级别日志
# - 每隔一段时间产生热更新日志
# - 模拟热更新后出现更多的错误日志
import os
import time
import random
import logging
def init_logging():
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger_format = logging.Formatter("[%(asctime)s][%(levelname)s][%(module)s:%(funcName)s:%(lineno)d] %(message)s")
logger_console = logging.StreamHandler()
logger_console.setLevel(logging.INFO)
logger_console.setFormatter(logger_format)
logger.addHandler(logger_console)
logger_file = logging.FileHandler("1.log")
logger_file.setLevel(logging.INFO)
logger_file.setFormatter(logger_format)
logger.addHandler(logger_file)
STAGE_CONFIG = {
"normal": {
"duration": "70±10",
"info_base_prob": 0.01,
"info_add_prob": 0,
"error_base_prob": 0.01,
"error_add_prob": 0,
},
"after_hotreload": {
"duration": "80±20",
"info_base_prob": 0.1,
"info_add_prob": 0.1,
"error_base_prob": 0.1,
"error_add_prob": 0.2,
}
}
INFO_LOG_TEMPLATE = [
"session connected",
"session closed",
"player login",
"player registered",
"player logout",
"buy item flow",
"sell item flow",
]
ERROR_LOG_TEMPLATE = [
"function %d error occur",
"player login failed",
"incorrect password",
"establish connection failed",
"bad arguments",
]
INFO_HOTRELOAD_TEMPLATE = [
"load file: %d.lua .......... [ok]"
]
def random_stage_time(duration):
pos = duration.find('±')
base_time = int(duration[:pos])
offset_time = int(duration[pos + 1:])
return base_time + (random.random() * 2 - 1) * offset_time
def random_select(l):
e = random.choice(l)
if e.find('%') >= 0:
r = random.randint(0, 10)
return e % r
return e
class App:
def __init__(self):
self._stage = None
self._stage_time = 0
self._timer = 0
self._init_state("normal")
def _init_state(self, stage):
self._stage = stage
self._stage_time = random_stage_time(STAGE_CONFIG[stage]["duration"])
self._timer = 0
def on_tick(self, elapsed_time):
self._timer += elapsed_time
if self._timer >= self._stage_time:
if self._stage == "normal":
logging.info(random_select(INFO_HOTRELOAD_TEMPLATE))
self._init_state("after_hotreload")
elif self._stage == "after_hotreload":
self._init_state("normal")
log_add_strength = 1 - min(1.0, (self._stage_time - self._timer) / self._stage_time)
log_info_prob = STAGE_CONFIG[self._stage]["info_base_prob"] + \
log_add_strength * STAGE_CONFIG[self._stage]["info_add_prob"]
log_err_prob = STAGE_CONFIG[self._stage]["error_base_prob"] + \
log_add_strength * STAGE_CONFIG[self._stage]["error_add_prob"]
if random.random() < log_info_prob:
logging.info(random_select(INFO_LOG_TEMPLATE))
if random.random() < log_err_prob:
logging.error(random_select(ERROR_LOG_TEMPLATE))
def main():
init_logging()
app = App()
last_time = time.time()
while True:
now_time = time.time()
app.on_tick(now_time - last_time)
last_time = now_time
time.sleep(0.01)
if __name__ == "__main__":
main()解决方法：
为了使用Python来实现上述要求，我们可以编写一个Python脚本来处理日志文件的读取、解析、分类统计以及比较热更新前后的错误日志数量。同时，我们将使用系统定时任务（如Linux下的cron）来定期运行这个Python脚本。
步骤 1: 编写Python脚本
首先，我们需要编写一个Python脚本来处理日志文件。这个脚本将需要能够：
- 读取日志文件。
- 检测热更新日志（以"load file:"开头）。
- 提取错误日志并根据内容进行分类统计。
- 比较热更新前后的错误日志数量。
- 记录超标的日志内容和统计信息到另一个文件。
下面是一个简化的Python脚本示例：
import re
import os
from datetime import datetime, timedelta
LOG_FILE = '/path/to/your/logfile.log'
OUTPUT_FILE = '/path/to/2.log'
TEMP_STATS_FILE = '/tmp/stats.tmp'
# 模拟从日志文件中读取行（实际中应使用文件操作）
def read_log_lines(file_path):
# 这里只是模拟，实际中应打开文件并逐行读取
with open(file_path, 'r') as file:
return file.readlines()
# 检测热更新并处理日志
def process_logs():
lines = read_log_lines(LOG_FILE)
update_time = None
before_counts = {}
after_counts = {}
# 遍历日志行
for line in lines:
if line.startswith('load file:'):
# 检测到热更新，记录时间并重置计数器
update_time = datetime.strptime(line.split(' ')[-1], '%Y-%m-%d %H:%M:%S')
before_counts = {}
after_counts = {}
elif 'ERROR' in line:
# 假设每行日志包含时间和级别
log_time = datetime.strptime(re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', line).group(), '%Y-%m-%d %H:%M:%S')
error_type = re.search(r'ERROR: (.+?)(?: \[|$)', line).group(1)
if update_time and update_time - timedelta(minutes=1) <= log_time <= update_time + timedelta(minutes=1):
# 在热更新前后1分钟内
if log_time < update_time:
before_counts[error_type] = before_counts.get(error_type, 0) + 1
else:
after_counts[error_type] = after_counts.get(error_type, 0) + 1
# 比较并记录结果
for type, after_count in after_counts.items():
before_count = before_counts.get(type, 0)
if after_count > before_count * 1.5:
with open(OUTPUT_FILE, 'a') as file:
file.write(f"Error type {type} increased significantly after update:\n")
file.write(f"Before: {before_count}, After: {after_count}\n")
# 这里可以添加更多逻辑来记录具体的日志内容
# 定时任务将调用这个函数
if __name__ == '__main__':
process_logs()注意：上面的脚本有几个简化和假设的地方，比如直接读取整个日志文件并假设时间戳和错误类型可以直接从日志行中提取。在实际应用中，你可能需要处理更复杂的日志格式和更大的日志文件。
步骤 2: 配置Cron定时任务
接下来，你需要在Linux系统上设置Cron定时任务来定期运行这个Python脚本。
- 打开终端。
- 输入
crontab -e命令来编辑当前用户的Cron任务列表。 - 添加一个定时任务来每分钟运行一次脚本（或者根据你的需求设置不同的频率）：
* * * * * /usr/bin/python3 /path/to/your/script.py注意将/usr/bin/python3替换为你的Python解释器的实际路径，/path/to/your/script.py替换为你的Python脚本的实际路径。
保存并关闭编辑器。Cron将自动加载新的任务列表。
现在，你的Python脚本将按照Cron任务的设置定期运行，检测日志文件中的热更新，并对错误
