Python

paramiko：SSH 批量管理基础 paramiko 是 Python 实现的 SSHv2 协议库，是运维自动化的底层基石。当需要精细控制 SSH 连接、处理非标准场景时，paramiko 提供了最大灵活性。 基础连接与命令执行 import paramiko import time def ssh_exec(host, port, username, password, command): """基础 SSH 命令执行""" client = paramiko.SSHClient() # 自动添加主机密钥（生产环境建议使用 known_hosts） client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(host, port=port, username=username, password=password, timeout=10) stdin, stdout, stderr = client.exec_command(command) # 获取退出码 exit_code = stdout.channel.recv_exit_status() output = stdout.read().decode().strip() error = stderr.read().decode().strip() return { 'host': host, 'exit_code': exit_code, 'output': output, 'error': error } finally: client.close() # 使用示例 result = ssh_exec('10....

概述 日志是系统运行时留下的最真实记录。当线上出问题时，日志是第一现场；当需要洞察系统行为时，日志是最丰富的数据源。但日志分析不是 grep 几个关键词那么简单——面对每天几十 GB 的日志，你需要高效的解析、灵活的聚合、智能的异常检测和清晰的可视化。本文用 Python 从零搭建一套完整的日志分析工具链。 参考来源：Python re 模块文档、pandas 文档 一、日志解析基础 1.1 正则表达式解析 日志解析的核心是把非结构化文本变成结构化数据。正则表达式是最基础也最灵活的工具： import re from datetime import datetime from typing import NamedTuple, Optional class LogEntry(NamedTuple): timestamp: datetime level: str message: str source: Optional[str] = None # 通用应用日志格式：2026-07-10 14:32:01 [ERROR] [app.payment] Payment failed: order=12345 APP_LOG_PATTERN = re.compile( r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+' r'\[(?P<level>DEBUG|INFO|WARN|WARNING|ERROR|FATAL)\]\s+' r'(?:\[(?P<source>[\w.]+)\]\s+)?' r'(?P<message>.+)' ) def parse_app_log(line: str) -> Optional[LogEntry]: """解析应用日志行""" match = APP_LOG_PATTERN.match(line.strip()) if not match: return None return LogEntry( timestamp=datetime....