Python自動化實現(xiàn)監(jiān)控分析系統(tǒng)日志并告警
更新時間：2025年12月05日 08:39:40 作者：零日失眠者
這篇文章主要為大家詳細介紹了如何使用Python實現(xiàn)一個強大的日志分析和告警工具,用于實時監(jiān)控、分析系統(tǒng)日志并發(fā)送告警通知,感興趣的小伙伴可以跟隨小編一起學習一下
功能介紹
這是一個強大的日志分析和告警工具，用于實時監(jiān)控、分析系統(tǒng)日志并發(fā)送告警通知。該工具具備以下核心功能：
多格式日志支持：
- 支持常見日志格式（Apache、Nginx、Syslog、JSON等）
- 自定義日志格式解析
- 實時日志流處理
- 歷史日志文件分析
智能模式匹配：
- 正則表達式模式匹配
- 關鍵詞和短語檢測
- 異常行為模式識別
- 統(tǒng)計分析和趨勢檢測
靈活告警機制：
- 多種告警渠道（郵件、短信、Slack、Webhook等）
- 告警級別和優(yōu)先級設置
- 告警抑制和去重
- 告警升級機制
實時監(jiān)控面板：
- 實時日志流可視化
- 關鍵指標儀表板
- 告警歷史查看
- 性能統(tǒng)計圖表
配置管理：
- YAML/JSON配置文件支持
- 動態(tài)規(guī)則加載和更新
- 多環(huán)境配置管理
- 規(guī)則模板和復用
場景應用
1. 系統(tǒng)運維監(jiān)控
- 監(jiān)控系統(tǒng)日志中的錯誤和異常
- 檢測安全威脅和入侵嘗試
- 跟蹤應用程序性能問題
- 自動響應系統(tǒng)故障
2. 安全事件檢測
- 檢測惡意登錄嘗試
- 識別可疑網(wǎng)絡活動
- 監(jiān)控文件訪問異常
- 實時安全威脅告警
3. 業(yè)務指標監(jiān)控
- 監(jiān)控用戶行為模式
- 檢測業(yè)務異常波動
- 跟蹤關鍵業(yè)務指標
- 自動生成業(yè)務報告
4. 合規(guī)性審計
- 監(jiān)控合規(guī)性相關日志
- 自動生成審計報告
- 檢測違規(guī)操作行為
- 滿足監(jiān)管要求
報錯處理
1. 日志文件訪問異常
try:
with open(log_file, 'r') as f:
process_log_lines(f)
except FileNotFoundError:
logger.error(f"日志文件不存在: {log_file}")
send_alert(f"日志文件缺失: {log_file}", level="CRITICAL")
except PermissionError:
logger.error(f"無權限訪問日志文件: {log_file}")
send_alert(f"日志文件訪問被拒絕: {log_file}", level="CRITICAL")
except IOError as e:
logger.error(f"讀取日志文件失敗: {str(e)}")
handle_io_error(log_file, e)
2. 日志解析異常
try:
parsed_log = parse_log_line(log_line)
if not parsed_log:
logger.warning(f"無法解析日志行: {log_line}")
increment_parse_error_count()
except LogParseError as e:
logger.error(f"日志解析錯誤: {str(e)}")
handle_parse_error(log_line, e)
except Exception as e:
logger.error(f"日志解析異常: {str(e)}")
handle_unexpected_parse_error(log_line, e)
3. 告警發(fā)送異常
try:
alert_sender.send(alert_message)
except AlertSendError as e:
logger.error(f"告警發(fā)送失敗: {str(e)}")
# 重試機制
retry_send_alert(alert_message, max_retries=3)
except NetworkError as e:
logger.error(f"網(wǎng)絡連接失敗: {str(e)}")
handle_network_failure(alert_message, e)
except Exception as e:
logger.error(f"告警發(fā)送異常: {str(e)}")
4. 配置文件異常
try:
config = load_config(config_file)
validate_config(config)
except yaml.YAMLError as e:
logger.error(f"配置文件YAML格式錯誤: {str(e)}")
raise ConfigError(f"配置文件格式無效: {str(e)}")
except json.JSONDecodeError as e:
logger.error(f"配置文件JSON格式錯誤: {str(e)}")
raise ConfigError(f"配置文件格式無效: {str(e)}")
except ValidationError as e:
logger.error(f"配置驗證失敗: {str(e)}")
raise ConfigError(f"配置無效: {str(e)}")
except Exception as e:
logger.error(f"配置加載異常: {str(e)}")
代碼實現(xiàn)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志分析和告警工具
功能：實時監(jiān)控、分析日志并發(fā)送告警
作者：Cline
版本：1.0
"""
import argparse
import sys
import json
import yaml
import logging
import os
import time
import threading
import re
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Callable
from collections import defaultdict, deque
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('log_analyzer.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class LogAnalyzerError(Exception):
"""日志分析器異常類"""
pass
class ConfigError(Exception):
"""配置異常類"""
pass
class AlertSendError(Exception):
"""告警發(fā)送異常類"""
pass
class LogEntry:
"""日志條目類"""
def __init__(self, raw_line: str, parsed_data: Dict = None):
self.raw_line = raw_line
self.parsed_data = parsed_data or {}
self.timestamp = parsed_data.get('timestamp') or datetime.now()
self.level = parsed_data.get('level', 'INFO')
self.source = parsed_data.get('source', 'unknown')
self.message = parsed_data.get('message', raw_line)
def to_dict(self):
"""轉(zhuǎn)換為字典"""
return {
'raw_line': self.raw_line,
'parsed_data': self.parsed_data,
'timestamp': self.timestamp.isoformat() if isinstance(self.timestamp, datetime) else self.timestamp,
'level': self.level,
'source': self.source,
'message': self.message
}
class LogParser:
"""日志解析器"""
def __init__(self, format_config: Dict):
self.format_config = format_config
self.pattern = re.compile(format_config.get('pattern', '.*'))
self.fields = format_config.get('fields', [])
def parse(self, log_line: str) -> Optional[LogEntry]:
"""解析日志行"""
try:
match = self.pattern.match(log_line.strip())
if not match:
return None
parsed_data = {}
for i, field in enumerate(self.fields):
if i < len(match.groups()):
parsed_data[field] = match.group(i + 1)
# 處理時間戳
if 'timestamp' in parsed_data:
try:
timestamp_str = parsed_data['timestamp']
# 嘗試多種時間格式
for fmt in [
'%Y-%m-%d %H:%M:%S',
'%d/%b/%Y:%H:%M:%S %z',
'%Y-%m-%dT%H:%M:%S',
'%Y-%m-%d %H:%M:%S.%f'
]:
try:
parsed_data['timestamp'] = datetime.strptime(timestamp_str, fmt)
break
except ValueError:
continue
except Exception:
pass
return LogEntry(log_line, parsed_data)
except Exception as e:
logger.warning(f"解析日志行失敗: {str(e)}")
return LogEntry(log_line)
class Rule:
"""告警規(guī)則"""
def __init__(self, rule_config: Dict):
self.id = rule_config.get('id')
self.name = rule_config.get('name', 'Unnamed Rule')
self.description = rule_config.get('description', '')
self.enabled = rule_config.get('enabled', True)
self.pattern = rule_config.get('pattern')
self.regex = re.compile(rule_config.get('regex', '.*')) if rule_config.get('regex') else None
self.level = rule_config.get('level', 'INFO')
self.threshold = rule_config.get('threshold', 1)
self.window = rule_config.get('window', 60) # 秒
self.cooldown = rule_config.get('cooldown', 300) # 秒
self.actions = rule_config.get('actions', [])
self.last_triggered = None
self.match_count = 0
self.match_history = deque(maxlen=1000) # 保留最近1000個匹配記錄
def should_trigger(self, log_entry: LogEntry) -> bool:
"""判斷是否應該觸發(fā)告警"""
if not self.enabled:
return False
# 檢查冷卻時間
if self.last_triggered and (datetime.now() - self.last_triggered).seconds < self.cooldown:
return False
# 檢查日志級別
if log_entry.level < self.level:
return False
# 檢查模式匹配
if self.pattern and self.pattern not in log_entry.message:
return False
# 檢查正則表達式
if self.regex and not self.regex.search(log_entry.message):
return False
# 記錄匹配
self.match_history.append({
'timestamp': datetime.now(),
'log_entry': log_entry.to_dict()
})
# 檢查閾值
if self.threshold > 1:
# 統(tǒng)計時間窗口內(nèi)的匹配次數(shù)
window_start = datetime.now() - timedelta(seconds=self.window)
recent_matches = [m for m in self.match_history
if m['timestamp'] >= window_start]
if len(recent_matches) < self.threshold:
return False
return True
def trigger(self, log_entry: LogEntry) -> Dict:
"""觸發(fā)告警"""
self.last_triggered = datetime.now()
self.match_count += 1
alert_data = {
'rule_id': self.id,
'rule_name': self.name,
'timestamp': datetime.now().isoformat(),
'log_entry': log_entry.to_dict(),
'match_count': self.match_count,
'recent_matches': list(self.match_history)[-10:] # 最近10個匹配
}
return alert_data
class AlertSender:
"""告警發(fā)送器"""
def __init__(self, config: Dict):
self.config = config
self.senders = {
'email': self._send_email,
'webhook': self._send_webhook,
'slack': self._send_slack,
'console': self._send_console
}
def send(self, alert_data: Dict, action_config: Dict):
"""發(fā)送告警"""
action_type = action_config.get('type', 'console')
sender = self.senders.get(action_type)
if not sender:
raise AlertSendError(f"不支持的告警類型: {action_type}")
try:
sender(alert_data, action_config)
except Exception as e:
raise AlertSendError(f"發(fā)送告警失敗: {str(e)}")
def _send_email(self, alert_data: Dict, config: Dict):
"""發(fā)送郵件告警"""
try:
msg = MIMEMultipart()
msg['From'] = config.get('sender')
msg['To'] = ', '.join(config.get('recipients', []))
msg['Subject'] = f"日志告警 - {alert_data['rule_name']}"
body = f"""
日志告警通知
規(guī)則名稱: {alert_data['rule_name']}
規(guī)則ID: {alert_data['rule_id']}
觸發(fā)時間: {alert_data['timestamp']}
匹配次數(shù): {alert_data['match_count']}
日志內(nèi)容:
{alert_data['log_entry']['raw_line']}
---
日志分析器
"""
msg.attach(MIMEText(body, 'plain'))
server = smtplib.SMTP(config.get('smtp_server'), config.get('smtp_port', 587))
server.starttls()
server.login(config.get('sender'), config.get('password'))
server.send_message(msg)
server.quit()
logger.info(f"郵件告警已發(fā)送: {alert_data['rule_name']}")
except Exception as e:
logger.error(f"發(fā)送郵件告警失敗: {str(e)}")
raise
def _send_webhook(self, alert_data: Dict, config: Dict):
"""發(fā)送Webhook告警"""
try:
response = requests.post(
config.get('url'),
json=alert_data,
headers=config.get('headers', {}),
timeout=config.get('timeout', 30)
)
response.raise_for_status()
logger.info(f"Webhook告警已發(fā)送: {alert_data['rule_name']}")
except Exception as e:
logger.error(f"發(fā)送Webhook告警失敗: {str(e)}")
raise
def _send_slack(self, alert_data: Dict, config: Dict):
"""發(fā)送Slack告警"""
try:
payload = {
'text': f"日志告警: {alert_data['rule_name']}",
'attachments': [{
'color': 'danger',
'fields': [
{
'title': '規(guī)則名稱',
'value': alert_data['rule_name'],
'short': True
},
{
'title': '觸發(fā)時間',
'value': alert_data['timestamp'],
'short': True
},
{
'title': '日志內(nèi)容',
'value': alert_data['log_entry']['raw_line'],
'short': False
}
]
}]
}
response = requests.post(
config.get('webhook_url'),
json=payload,
timeout=config.get('timeout', 30)
)
response.raise_for_status()
logger.info(f"Slack告警已發(fā)送: {alert_data['rule_name']}")
except Exception as e:
logger.error(f"發(fā)送Slack告警失敗: {str(e)}")
raise
def _send_console(self, alert_data: Dict, config: Dict):
"""發(fā)送控制臺告警"""
print(f"[ALERT] {alert_data['rule_name']}: {alert_data['log_entry']['message']}")
class LogFileMonitor:
"""日志文件監(jiān)控器"""
def __init__(self, file_path: str, parser: LogParser):
self.file_path = file_path
self.parser = parser
self.file_handle = None
self.file_position = 0
self.callbacks = []
def add_callback(self, callback: Callable):
"""添加回調(diào)函數(shù)"""
self.callbacks.append(callback)
def start(self):
"""開始監(jiān)控"""
try:
# 打開文件并移動到末尾
self.file_handle = open(self.file_path, 'r', encoding='utf-8', errors='ignore')
self.file_handle.seek(0, 2) # 移動到文件末尾
self.file_position = self.file_handle.tell()
# 啟動監(jiān)控線程
thread = threading.Thread(target=self._monitor_loop)
thread.daemon = True
thread.start()
logger.info(f"開始監(jiān)控日志文件: {self.file_path}")
except Exception as e:
logger.error(f"啟動日志監(jiān)控失敗: {str(e)}")
raise LogAnalyzerError(f"啟動監(jiān)控失敗: {str(e)}")
def stop(self):
"""停止監(jiān)控"""
if self.file_handle:
self.file_handle.close()
self.file_handle = None
logger.info(f"停止監(jiān)控日志文件: {self.file_path}")
def _monitor_loop(self):
"""監(jiān)控循環(huán)"""
while self.file_handle:
try:
# 檢查文件是否有新內(nèi)容
current_position = self.file_handle.tell()
self.file_handle.seek(0, 2)
file_size = self.file_handle.tell()
self.file_handle.seek(current_position)
if file_size > current_position:
# 讀取新內(nèi)容
new_lines = self.file_handle.readlines()
for line in new_lines:
self._process_line(line)
# 更新文件位置
self.file_position = self.file_handle.tell()
time.sleep(0.1) # 短暫休眠
except Exception as e:
logger.error(f"日志監(jiān)控異常: {str(e)}")
time.sleep(1)
def _process_line(self, line: str):
"""處理日志行"""
try:
log_entry = self.parser.parse(line)
if log_entry:
for callback in self.callbacks:
try:
callback(log_entry)
except Exception as e:
logger.error(f"回調(diào)執(zhí)行失敗: {str(e)}")
except Exception as e:
logger.warning(f"處理日志行失敗: {str(e)}")
class DatabaseManager:
"""數(shù)據(jù)庫管理器"""
def __init__(self, db_path: str = 'log_analyzer.db'):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化數(shù)據(jù)庫"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 創(chuàng)建告警記錄表
cursor.execute('''
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
rule_id TEXT NOT NULL,
rule_name TEXT NOT NULL,
log_entry TEXT,
match_count INTEGER
)
''')
# 創(chuàng)建統(tǒng)計信息表
cursor.execute('''
CREATE TABLE IF NOT EXISTS statistics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
rule_id TEXT NOT NULL,
match_count INTEGER,
processed_lines INTEGER
)
''')
conn.commit()
conn.close()
logger.info("數(shù)據(jù)庫初始化完成")
except Exception as e:
logger.error(f"數(shù)據(jù)庫初始化失敗: {str(e)}")
def save_alert(self, alert_data: Dict):
"""保存告警記錄"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO alerts
(timestamp, rule_id, rule_name, log_entry, match_count)
VALUES (?, ?, ?, ?, ?)
''', (
alert_data.get('timestamp', ''),
alert_data.get('rule_id', ''),
alert_data.get('rule_name', ''),
json.dumps(alert_data.get('log_entry', {})),
alert_data.get('match_count', 0)
))
conn.commit()
conn.close()
logger.info(f"告警記錄已保存: {alert_data.get('rule_name', '')}")
except Exception as e:
logger.error(f"保存告警記錄失敗: {str(e)}")
def get_alerts(self, limit: int = 100) -> List[Dict]:
"""獲取告警記錄"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT timestamp, rule_id, rule_name, log_entry, match_count
FROM alerts
ORDER BY timestamp DESC
LIMIT ?
''', (limit,))
rows = cursor.fetchall()
conn.close()
alerts = []
for row in rows:
alerts.append({
'timestamp': row[0],
'rule_id': row[1],
'rule_name': row[2],
'log_entry': json.loads(row[3]) if row[3] else {},
'match_count': row[4]
})
return alerts
except Exception as e:
logger.error(f"獲取告警記錄失敗: {str(e)}")
return []
class LogAnalyzer:
"""日志分析器主類"""
def __init__(self, config_file: str = None):
self.config_file = config_file
self.config = {}
self.rules = []
self.parsers = {}
self.monitors = []
self.alert_sender = None
self.db_manager = None
self.running = False
self.processed_lines = 0
self.match_statistics = defaultdict(int)
# 加載配置
self.load_config()
# 初始化組件
self._init_components()
def load_config(self):
"""加載配置文件"""
if not self.config_file or not os.path.exists(self.config_file):
logger.info("未指定配置文件或文件不存在，使用默認配置")
self.config = self._create_default_config()
return
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
if self.config_file.endswith('.yaml') or self.config_file.endswith('.yml'):
self.config = yaml.safe_load(f)
else:
self.config = json.load(f)
logger.info(f"成功加載配置文件: {self.config_file}")
except Exception as e:
logger.error(f"加載配置文件失敗: {str(e)}")
raise ConfigError(f"配置加載失敗: {str(e)}")
def _create_default_config(self) -> Dict:
"""創(chuàng)建默認配置"""
return {
"parsers": {
"default": {
"pattern": "(.*)",
"fields": ["message"]
}
},
"rules": [
{
"id": "error_detector",
"name": "錯誤檢測器",
"description": "檢測包含ERROR關鍵字的日志",
"enabled": True,
"pattern": "ERROR",
"level": "ERROR",
"threshold": 1,
"window": 60,
"cooldown": 300,
"actions": [
{"type": "console"}
]
}
],
"sources": [
{
"path": "/var/log/syslog",
"parser": "default",
"enabled": False
}
],
"actions": {
"email": {
"type": "email",
"sender": "loganalyzer@example.com",
"password": "your_password",
"smtp_server": "smtp.example.com",
"smtp_port": 587,
"recipients": ["admin@example.com"]
}
}
}
def _init_components(self):
"""初始化組件"""
# 初始化解析器
for name, parser_config in self.config.get('parsers', {}).items():
self.parsers[name] = LogParser(parser_config)
# 初始化規(guī)則
for rule_config in self.config.get('rules', []):
self.rules.append(Rule(rule_config))
# 初始化告警發(fā)送器
self.alert_sender = AlertSender(self.config.get('actions', {}))
# 初始化數(shù)據(jù)庫管理器
self.db_manager = DatabaseManager()
def start(self):
"""啟動日志分析器"""
if self.running:
logger.warning("日志分析器已在運行")
return
logger.info("啟動日志分析器...")
self.running = True
# 啟動日志監(jiān)控器
for source_config in self.config.get('sources', []):
if not source_config.get('enabled', False):
continue
path = source_config.get('path')
parser_name = source_config.get('parser', 'default')
if not os.path.exists(path):
logger.warning(f"日志文件不存在: {path}")
continue
if parser_name not in self.parsers:
logger.warning(f"解析器不存在: {parser_name}")
continue
try:
parser = self.parsers[parser_name]
monitor = LogFileMonitor(path, parser)
monitor.add_callback(self._process_log_entry)
monitor.start()
self.monitors.append(monitor)
logger.info(f"已啟動日志監(jiān)控: {path}")
except Exception as e:
logger.error(f"啟動日志監(jiān)控失敗 {path}: {str(e)}")
logger.info("日志分析器啟動完成")
def stop(self):
"""停止日志分析器"""
logger.info("停止日志分析器...")
self.running = False
# 停止所有監(jiān)控器
for monitor in self.monitors:
try:
monitor.stop()
except Exception as e:
logger.error(f"停止監(jiān)控器失敗: {str(e)}")
self.monitors.clear()
logger.info("日志分析器已停止")
def _process_log_entry(self, log_entry: LogEntry):
"""處理日志條目"""
self.processed_lines += 1
# 應用所有規(guī)則
for rule in self.rules:
try:
if rule.should_trigger(log_entry):
# 觸發(fā)告警
alert_data = rule.trigger(log_entry)
# 保存到數(shù)據(jù)庫
if self.db_manager:
self.db_manager.save_alert(alert_data)
# 更新統(tǒng)計信息
self.match_statistics[rule.id] += 1
# 發(fā)送告警
for action_config in rule.actions:
try:
self.alert_sender.send(alert_data, action_config)
except Exception as e:
logger.error(f"發(fā)送告警失敗: {str(e)}")
except Exception as e:
logger.error(f"應用規(guī)則 {rule.name} 失敗: {str(e)}")
def get_statistics(self) -> Dict:
"""獲取統(tǒng)計信息"""
return {
'processed_lines': self.processed_lines,
'match_statistics': dict(self.match_statistics),
'active_monitors': len(self.monitors),
'active_rules': len([r for r in self.rules if r.enabled])
}
def add_rule(self, rule_config: Dict):
"""動態(tài)添加規(guī)則"""
try:
rule = Rule(rule_config)
self.rules.append(rule)
logger.info(f"已添加規(guī)則: {rule.name}")
except Exception as e:
logger.error(f"添加規(guī)則失敗: {str(e)}")
def remove_rule(self, rule_id: str):
"""移除規(guī)則"""
self.rules = [r for r in self.rules if r.id != rule_id]
logger.info(f"已移除規(guī)則: {rule_id}")
def get_alerts(self, limit: int = 100) -> List[Dict]:
"""獲取告警記錄"""
if self.db_manager:
return self.db_manager.get_alerts(limit)
return []
def create_sample_config():
"""創(chuàng)建示例配置文件"""
sample_config = {
"parsers": {
"nginx": {
"pattern": r'(\S+) - - \[(.*?)\] "(\S+) (\S+) (\S+)" (\d+) (\d+) "(.*?)" "(.*?)"',
"fields": ["ip", "timestamp", "method", "url", "protocol", "status", "size", "referer", "user_agent"]
},
"syslog": {
"pattern": r'<(\d+)>(\d+) (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+[+-]\d{2}:\d{2}) (\S+) ([^ ]+) ([^ ]+) (.*)',
"fields": ["priority", "version", "timestamp", "hostname", "app_name", "proc_id", "message"]
}
},
"rules": [
{
"id": "nginx_404_errors",
"name": "Nginx 404錯誤檢測",
"description": "檢測Nginx訪問日志中的404錯誤",
"enabled": True,
"regex": r'" 404 ',
"level": "WARNING",
"threshold": 5,
"window": 60,
"cooldown": 300,
"actions": [
{"type": "console"},
{"type": "email"}
]
},
{
"id": "security_login_failures",
"name": "安全登錄失敗檢測",
"description": "檢測認證失敗的日志",
"enabled": True,
"pattern": "authentication failure",
"level": "ERROR",
"threshold": 3,
"window": 300,
"cooldown": 600,
"actions": [
{"type": "console"},
{"type": "webhook", "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"}
]
}
],
"sources": [
{
"path": "/var/log/nginx/access.log",
"parser": "nginx",
"enabled": False
},
{
"path": "/var/log/syslog",
"parser": "syslog",
"enabled": False
}
],
"actions": {
"email": {
"type": "email",
"sender": "loganalyzer@example.com",
"password": "your_app_password",
"smtp_server": "smtp.gmail.com",
"smtp_port": 587,
"recipients": ["admin@example.com", "security@example.com"]
},
"slack_webhook": {
"type": "webhook",
"url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
"timeout": 30
}
}
}
with open('log_analyzer_sample_config.json', 'w', encoding='utf-8') as f:
json.dump(sample_config, f, indent=2, ensure_ascii=False)
logger.info("示例配置文件已創(chuàng)建: log_analyzer_sample_config.json")
def main():
parser = argparse.ArgumentParser(description='日志分析和告警工具')
parser.add_argument('-c', '--config', help='配置文件路徑')
parser.add_argument('--start', action='store_true', help='啟動日志分析器')
parser.add_argument('--sample-config', action='store_true', help='創(chuàng)建示例配置文件')
parser.add_argument('--stats', action='store_true', help='顯示統(tǒng)計信息')
parser.add_argument('--alerts', action='store_true', help='顯示最近告警')
parser.add_argument('--limit', type=int, default=10, help='告警記錄數(shù)量限制')
args = parser.parse_args()
if args.sample_config:
create_sample_config()
return
analyzer = LogAnalyzer(args.config)
if args.stats:
stats = analyzer.get_statistics()
print(json.dumps(stats, indent=2, ensure_ascii=False))
return
if args.alerts:
alerts = analyzer.get_alerts(args.limit)
print(json.dumps(alerts, indent=2, ensure_ascii=False))
return
if args.start:
try:
analyzer.start()
# 保持程序運行
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("收到中斷信號，正在停止分析器...")
finally:
analyzer.stop()
else:
parser.print_help()
if __name__ == '__main__':
main()
使用說明
1. 安裝依賴
pip install pyyaml requests
2. 創(chuàng)建配置文件
python log_analyzer.py --sample-config
3. 啟動日志分析器
python log_analyzer.py --config log_analyzer_config.json --start
4. 查看統(tǒng)計信息
python log_analyzer.py --stats
5. 查看告警記錄
python log_analyzer.py --alerts --limit 20
配置文件示例
JSON配置文件
{
"parsers": {
"nginx": {
"pattern": "(\\S+) - - \\[(.*?)\\] \"(\\S+) (\\S+) (\\S+)\" (\\d+) (\\d+) \"(.*?)\" \"(.*?)\"",
"fields": ["ip", "timestamp", "method", "url", "protocol", "status", "size", "referer", "user_agent"]
},
"syslog": {
"pattern": "<(\\d+)>(\\d+) (\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d+[+-]\\d{2}:\\d{2}) (\\S+) ([^ ]+) ([^ ]+) (.*)",
"fields": ["priority", "version", "timestamp", "hostname", "app_name", "proc_id", "message"]
}
},
"rules": [
{
"id": "nginx_404_errors",
"name": "Nginx 404錯誤檢測",
"description": "檢測Nginx訪問日志中的404錯誤",
"enabled": true,
"regex": "\" 404 ",
"level": "WARNING",
"threshold": 5,
"window": 60,
"cooldown": 300,
"actions": [
{"type": "console"},
{"type": "email"}
]
},
{
"id": "security_login_failures",
"name": "安全登錄失敗檢測",
"description": "檢測認證失敗的日志",
"enabled": true,
"pattern": "authentication failure",
"level": "ERROR",
"threshold": 3,
"window": 300,
"cooldown": 600,
"actions": [
{"type": "console"},
{"type": "webhook", "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"}
]
}
],
"sources": [
{
"path": "/var/log/nginx/access.log",
"parser": "nginx",
"enabled": false
},
{
"path": "/var/log/syslog",
"parser": "syslog",
"enabled": false
}
],
"actions": {
"email": {
"type": "email",
"sender": "loganalyzer@example.com",
"password": "your_app_password",
"smtp_server": "smtp.gmail.com",
"smtp_port": 587,
"recipients": ["admin@example.com", "security@example.com"]
},
"slack_webhook": {
"type": "webhook",
"url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
"timeout": 30
}
}
}
高級特性
1. 智能模式匹配
支持正則表達式和關鍵詞匹配，能夠識別復雜的日志模式和異常行為。
2. 告警抑制機制
通過閾值和冷卻時間設置，避免重復告警和告警風暴。
3. 多渠道告警
支持郵件、Webhook、Slack等多種告警渠道，確保告警及時送達。
4. 歷史數(shù)據(jù)分析
內(nèi)置數(shù)據(jù)庫存儲告警記錄和統(tǒng)計信息，支持歷史數(shù)據(jù)分析和趨勢預測。
最佳實踐
1. 規(guī)則配置優(yōu)化
- 根據(jù)實際需求設置合理的閾值和時間窗口
- 使用具體的正則表達式而非通用模式以提高性能
- 定期審查和優(yōu)化告警規(guī)則
2. 性能調(diào)優(yōu)
- 合理設置日志監(jiān)控文件的數(shù)量和大小
- 使用高效的正則表達式模式
- 定期清理歷史數(shù)據(jù)
3. 安全性考慮
- 保護配置文件中的敏感信息
- 限制對日志文件的訪問權限
- 定期審查告警記錄
總結
這個日志分析和告警工具提供了一個功能強大、靈活可配置的日志監(jiān)控解決方案。通過實時分析日志數(shù)據(jù)并發(fā)送及時告警，可以幫助運維人員快速發(fā)現(xiàn)和響應系統(tǒng)問題，提高系統(tǒng)的穩(wěn)定性和安全性。
到此這篇關于Python自動化實現(xiàn)監(jiān)控分析系統(tǒng)日志并告警的文章就介紹到這了,更多相關Python日志監(jiān)控分析內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Django開發(fā)web后端對比SpringBoot示例分析
這篇文章主要介紹了Django開發(fā)web后端對比SpringBoot示例分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-12-12
python之pyqt5通過按鈕改變Label的背景顏色方法
今天小編就為大家分享一篇python之pyqt5通過按鈕改變Label的背景顏色方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-06-06
這篇文章主要介紹了python類方法和靜態(tài)方法,實例分析了Python中返回一個返回值與多個返回值的方法,需要的朋友可以參考下2021-10-10
這篇文章主要介紹了python多線程與多進程及其區(qū)別詳解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2021-08-08
如何實現(xiàn)python爬蟲爬取視頻時實現(xiàn)實時進度條顯示
這篇文章主要介紹了如何實現(xiàn)python爬蟲爬取視頻時實現(xiàn)實時進度條顯示，在爬取并下載網(wǎng)頁上的視頻的時候，我們需要實時進度條，這可以幫助我們更直觀的看到視頻的下載進度。文章圍繞主題展開更多內(nèi)容，需要的小伙伴可以參考一下2022-06-06
keras中的loss、optimizer、metrics用法
這篇文章主要介紹了keras中的loss、optimizer、metrics用法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06
Python實現(xiàn)的統(tǒng)計文章單詞次數(shù)功能示例
這篇文章主要介紹了Python實現(xiàn)的統(tǒng)計文章單詞次數(shù)功能,結合實例形式分析了Python針對字符串序列的遍歷、計算等相關操作技巧,需要的朋友可以參考下2019-07-07
這篇文章主要介紹了python安裝gdal的兩種方法,每種方法給大家介紹的都非常詳細，具有一定的參考借鑒價值，需要的朋友可以參考下2019-10-10
OpenCV python sklearn隨機超參數(shù)搜索的實現(xiàn)
這篇文章主要介紹了OpenCV python sklearn隨機超參數(shù)搜索的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-01-01
python Tcp協(xié)議發(fā)送和接收信息的例子
今天小編就為大家分享一篇python Tcp協(xié)議發(fā)送和接收信息的例子，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07