class LogFollower:
"""类似 tail -f 的Python实现，支持日志轮转"""
def __init__(self, log_file):
self.log_file = log_file
self._file = None
self._inode = None
self._buffer = deque(maxlen=100) # 保留最近100行用于上下文

def _get_inode(self):
return os.stat(self.log_file).st_ino

def _open_file(self):
if self._file:
self._file.close()
self._file = open(self.log_file, 'r')
self._file.seek(0,

2) # 跳到文件末尾
self._inode = self._get_inode()

def follow(self):
"""持续追踪日志文件，自动处理轮转"""
self._open_file()
while True:
# 检查文件是否被轮转（inode变了）
current_inode = self._get_inode()
if current_inode != self._inode:
self._open_file()
continue

第二步：日志解析——把字符串变成结构化数据

有了原始日志，接下来就是解析。这一步最坑。因为不同应用的日志格式千奇百怪，而且官方文档经常文档不够清晰。

我的原则是：先统一格式，再分析。不管你用什么日志框架，最终都给我变成JSON或者固定的分隔符格式。

这里给一个解析标准Nginx日志的示例：

Nginx combined格式示例：

192.168.1.1 - - [28/Feb/2024:10:15:30 +0800] "GET /api/users HTTP/1.1" 200 1234 "-" "Mozilla/5.0"

NGINX_PATTERN = re.compile(
r'(?P\S+) \S+ \S+ \[(?P[^\]]+)\] '
r'"(?P\S+) (?P\S+) \S+" '
r'(?P\d{3}) (?P\d+) '
r'"(?P[^"]*)" "(?P[^"]*)"'
)

def parse_nginx_line(line: str) -> Optional[Dict]:
"""将一行Nginx日志解析为结构化数据"""
match = NGINX_PATTERN.match(line)
if not match:
return None

data = match.groupdict()

# 解析时间到Python datetime对象
try:
data['time'] = datetime.strptime(
data['time'],
'%d/%b/%Y:%H:%M:%S %z'
)
except ValueError:
data['time'] = None

# 类型转换
data['status'] = int(data['status'])
data['size'] = int(data['size'])

return data

使用示例

优化的解析方式：先粗筛，再精解

第三步：实时告警——别让问题过夜

解析完日志，我们要做的是实时告警。我踩过的坑是：告警太多等于没有告警。一开始我设置了每出现一个5xx就发一次告警，结果群里的告警消息刷屏，大家直接静默了。

正确的做法是：聚合告警 + 分级告警。

class LogAlert:
"""基于滑动窗口的日志告警系统"""

def __init__(self, redis_host='localhost', redis_port=6379):
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.window_size = 60 # 60秒滑动窗口
self.thresholds = {
'critical': 100, # 60秒内100个5xx → 严重
'warning': 20, # 60秒内20个5xx → 警告
}
self.webhook_url = 'https://your-webhook-url.com/alert'

def should_alert(self, error_type: str, count: int) -> str:
"""根据错误次数决定告警级别"""
if count >= self.thresholds['critical']:
return 'critical'
elif count >= self.thresholds['warning']:
return 'warning'
return 'info'

def process_error(self, log_entry: Dict):
"""处理一条错误日志，使用Redis的Sorted Set做滑动窗口计数"""
now = datetime.now()
window_key = f"errors:{log_entry['path']}:{now.strftime('%Y%m%d%H%M')}"

# 将当前错误加入窗口，用时间戳作为score
self.redis.zadd(window_key, {json.dumps(log_entry): now.timestamp()})

# 清理60秒前的数据
cutoff = now.timestamp() - self.window_size
self.redis.zremrangebyscore(window_key, '-inf', cutoff)

# 统计当前窗口内的错误数
error_count = self.redis.zcard(window_key)

# 判断是否需要告警
level = self.should_alert(log_entry['path'], error_count)
if level in ('critical', 'warning'):
self.send_alert(level, log_entry, error_count)

def send_alert(self, level: str, log_entry: Dict, count: int):
"""发送告警到企业微信/钉钉/Slack"""
message = {
'level': level,
'path': log_entry['path'],
'ip': log_entry['ip'],
'count': count,
'time': log_entry['time'].isoformat(),
'message': f"[{level.upper()}] 路径 {log_entry['path']} 在60秒内出现 {count} 次错误"
}

核心配图：一张Redis Sorted Set的数据结构示意图，展示如何用时间戳做滑动窗口计数。

第四步：可视化——让数据自己说话

告警是救火，可视化才是防火。我把所有解析后的日志写入InfluxDB（时序数据库），然后用Grafana画看板。这样你能看到：

• 哪个API在哪个时段最慢
• 错误分布的地域/IP特征
• 系统负载和错误数的关联

这里给一个最简化的实现——不用InfluxDB，直接用Python往CSV写，然后用Pandas做实时图表：

class LogVisualizer:
"""简易的实时日志可视化"""

def __init__(self, max_points=1000):
self.data = deque(maxlen=max_points)
self.lock = threading.Lock()

def add_point(self, timestamp: datetime, metric: str, value: float):
"""添加一个数据点"""
with self.lock:
self.data.append({
'timestamp': timestamp,
'metric': metric,
'value': value
})

def get_dataframe(self) -> pd.DataFrame:
"""将deque转为DataFrame用于分析"""
with self.lock:
df = pd.DataFrame(self.data)
if not df.empty:
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 按分钟聚合
df.set_index('timestamp', inplace=True)
return df.resample('1T').mean()
return pd.DataFrame()

def plot_realtime(self):
"""实时更新图表（每5秒刷新一次）"""
plt.ion()
fig, ax = plt.subplots(figsize=(12, 6))

还有个技巧：不是所有日志都需要可视化。我只把错误率、响应时间、QPS这三个指标画出来，太多信息反而干扰判断。记住二八法则——20%的指标决定了80%的问题。

总结前配图：一张Grafana看板的截图，展示错误率、响应延迟、QPS三个核心指标随时间变化的曲线。

总结一下，你可以立刻用的三个点