1. 项目概述日志监控与警报系统的核心价值服务器日志就像系统的黑匣子记录着每一次请求、每一个错误和每一条关键事件。但海量的日志数据如果不加以监控就如同把金矿埋在地下——直到系统崩溃时才会发现那些早已被记录下来的预警信号。这就是为什么我们需要自动化日志监控系统。这个Python项目解决的核心问题是实时扫描系统日志文件识别预设的关键错误模式如ERROR、CRITICAL等并通过邮件或即时消息平台发送警报通知。相比商业监控工具自建方案的优势在于完全定制化可以针对特定应用定义专属的错误模式零成本利用现有服务器资源无需额外订阅费用深度集成能与内部系统无缝对接不受SaaS产品功能限制2. 技术选型与架构设计2.1 为什么选择PythonPython在日志处理领域有三大不可替代的优势丰富的文本处理库re模块提供强大的正则表达式支持能高效匹配复杂日志模式成熟的邮件发送方案smtplibemail库组合可以处理各种邮件格式和附件跨平台文件监控watchdog库能可靠地检测文件变化兼容Linux/Windows系统2.2 系统架构设计典型的日志监控系统包含以下组件日志文件 → 文件监控 → 模式匹配 → 警报触发 → 通知发送我们采用生产者-消费者模型生产者线程使用watchdog监控日志目录变化消费者线程用正则表达式匹配关键错误通过SMTP发送邮件3. 核心实现步骤详解3.1 环境准备与依赖安装首先确保Python 3.6环境然后安装必要依赖pip install watchdog python-dotenv提示使用python-dotenv管理敏感配置如邮件密码不要硬编码在脚本中3.2 日志监控模块实现使用watchdog的FileSystemEventHandler类创建自定义处理器from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogHandler(FileSystemEventHandler): def on_modified(self, event): if not event.is_directory and event.src_path.endswith(.log): with open(event.src_path) as f: new_lines f.readlines()[-10:] # 读取最后10行新日志 analyze_logs(new_lines)3.3 日志分析引擎核心的正则匹配逻辑示例import re ERROR_PATTERNS [ rERROR.*(timeout|failed), rCRITICAL.*disk\sfull, rAlert:\sCPU\soverload ] def analyze_logs(lines): for line in lines: for pattern in ERROR_PATTERNS: if re.search(pattern, line, re.IGNORECASE): send_alert(f匹配到错误: {pattern}\n日志内容: {line}) break3.4 邮件警报系统使用smtplib发送HTML格式警报邮件import smtplib from email.mime.text import MIMEText from email.header import Header def send_alert(content): msg MIMEText(content, html, utf-8) msg[Subject] Header( 系统异常警报, utf-8) msg[From] monitoryourdomain.com msg[To] adminyourdomain.com with smtplib.SMTP(smtp.server.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)4. 高级功能扩展4.1 频率限制与告警聚合避免短时间内重复发送相同错误的警报from collections import defaultdict from datetime import datetime, timedelta alert_history defaultdict(list) def should_send_alert(error_type): now datetime.now() # 相同错误1小时内不重复发送 recent_alerts [t for t in alert_history[error_type] if now - t timedelta(hours1)] alert_history[error_type].append(now) return len(recent_alerts) 04.2 多通知渠道集成除了邮件还可以添加Slack/webhook支持import requests def send_to_slack(message): webhook_url https://hooks.slack.com/services/XXX payload {text: message} requests.post(webhook_url, jsonpayload)5. 生产环境部署方案5.1 系统服务化部署使用systemd管理监控进程Linux环境# /etc/systemd/system/logmonitor.service [Unit] DescriptionLog Monitor Service [Service] ExecStart/usr/bin/python3 /opt/logmonitor/main.py Restartalways Userroot [Install] WantedBymulti-user.target5.2 日志轮转处理处理logrotate场景的完整方案def handle_rotated_file(file_path): # 检查文件是否被轮转inode变化 current_inode os.stat(file_path).st_ino if current_inode ! getattr(handle_rotated_file, last_inode, None): handle_rotated_file.last_inode current_inode # 重新打开文件读取最新内容 with open(file_path) as f: analyze_logs(f.readlines()[-100:]) # 读取轮转后文件末尾6. 性能优化技巧6.1 高效文件读取方案避免重复读取整个文件def tail(filename, n10): 返回文件最后n行 with open(filename, rb) as f: # 从文件末尾开始读取 f.seek(0, 2) end f.tell() lines_found 0 offset 1 while lines_found n and offset end: f.seek(max(0, end - offset)) data f.read(min(1024, offset)) lines_found data.count(b\n) offset * 2 f.seek(max(0, end - offset)) return f.readlines()[-n:]6.2 正则表达式优化预编译正则模式提升匹配速度COMPILED_PATTERNS [re.compile(p, re.IGNORECASE) for p in ERROR_PATTERNS] def analyze_logs(lines): for line in lines: for pattern in COMPILED_PATTERNS: if pattern.search(line): # ...警报逻辑7. 常见问题与解决方案7.1 文件权限问题典型错误PermissionError: [Errno 13] Permission denied解决方案确保运行用户有日志文件读取权限对于docker容器需要正确挂载volumedocker run -v /var/log:/host_logs logmonitor7.2 字符编码问题处理不同编码的日志文件def try_decode(line): for encoding in [utf-8, gbk, latin-1]: try: return line.decode(encoding) except UnicodeDecodeError: continue return line.decode(utf-8, errorsreplace)7.3 邮件发送失败SMTP常见问题排查检查防火墙是否开放587端口验证是否开启SMTP认证测试Telnet连接telnet smtp.server.com 5878. 监控指标与可视化8.1 关键指标收集记录监控系统自身运行状态monitor_metrics { files_processed: 0, errors_found: 0, last_alert: None } def export_metrics(): return ( f# HELP logmonitor_errors Total errors detected\n f# TYPE logmonitor_errors counter\n flogmonitor_errors {monitor_metrics[errors_found]}\n )8.2 Prometheus集成暴露监控指标端点from http.server import HTTPServer, BaseHTTPRequestHandler class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /metrics: self.send_response(200) self.end_headers() self.wfile.write(export_metrics().encode()) HTTPServer((0.0.0.0, 8000), MetricsHandler).serve_forever()9. 安全最佳实践9.1 敏感信息处理避免日志中的密码泄露SENSITIVE_KEYS [password, secret, api_key] def sanitize_log(line): for key in SENSITIVE_KEYS: line re.sub(fr{key}[^\s], f{key}[REDACTED], line) return line9.2 访问控制限制监控目录范围ALLOWED_PATHS [/var/log/app, /tmp/logs] def is_allowed(path): return any(path.startswith(allowed) for allowed in ALLOWED_PATHS)10. 测试方案设计10.1 单元测试用例测试日志分析核心逻辑import unittest class TestLogAnalysis(unittest.TestCase): def test_error_detection(self): test_lines [ INFO: System started, ERROR: Database connection failed, CRITICAL: Disk space exhausted ] with self.assertLogs() as cm: analyze_logs(test_lines) self.assertIn(ERROR, cm.output[0])10.2 集成测试方案模拟完整工作流程from unittest.mock import patch def test_integration(): with patch(smtplib.SMTP) as mock_smtp: # 触发文件修改事件 event type(, (), {is_directory: False, src_path: test.log})() handler LogHandler() handler.on_modified(event) # 验证邮件发送被调用 assert mock_smtp.return_value.send_message.called11. 性能基准测试使用不同日志量测试处理速度import timeit def benchmark(): setup from __main__ import analyze_logs test_data [ERROR: test] * 10000 stmt analyze_logs(test_data) return timeit.timeit(stmt, setup, number100) print(f处理速度: {benchmark():.2f}秒/万条)12. 容器化部署Dockerfile最佳实践FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [python, main.py]构建与运行docker build -t logmonitor . docker run -d -v /var/log:/logs logmonitor13. 配置管理方案使用YAML文件管理配置# config.yaml monitor: paths: - /var/log/nginx - /opt/app/logs patterns: - ERROR - CRITICAL alert: email: server: smtp.example.com recipients: - adminexample.comPython读取配置import yaml with open(config.yaml) as f: config yaml.safe_load(f)14. 高可用设计14.1 心跳检测机制实现监控进程的健康检查import threading def heart_beat(): while True: with open(/tmp/logmonitor.heartbeat, w) as f: f.write(str(time.time())) time.sleep(60) threading.Thread(targetheart_beat, daemonTrue).start()14.2 分布式监控方案使用Redis实现多节点协同import redis r redis.Redis() def acquire_lock(lock_name, expire300): return r.set(lock_name, 1, nxTrue, exexpire) if acquire_lock(logmonitor:master): # 当前节点获得master角色 start_monitoring()15. 日志分析算法进阶15.1 异常检测算法基于统计的异常值检测import numpy as np def detect_anomaly(error_counts, window24): 使用Z-score检测异常错误量 avg np.mean(error_counts[-window:]) std np.std(error_counts[-window:]) latest error_counts[-1] return abs(latest - avg) 3 * std15.2 机器学习分类使用scikit-learn进行日志分类from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression vectorizer TfidfVectorizer() classifier LogisticRegression() # 训练样本示例 train_texts [ERROR: db fail, INFO: system ok] train_labels [1, 0] X vectorizer.fit_transform(train_texts) classifier.fit(X, train_labels) def classify_log(line): return classifier.predict(vectorizer.transform([line]))16. 系统资源管理16.1 内存限制方案防止日志文件过大导致OOMimport resource def set_memory_limit(percent0.8): soft, hard resource.getrlimit(resource.RLIMIT_AS) total_mem os.sysconf(SC_PAGE_SIZE) * os.sysconf(SC_PHYS_PAGES) new_limit int(total_mem * percent) resource.setrlimit(resource.RLIMIT_AS, (new_limit, hard))16.2 CPU节流控制限制日志分析CPU占用import psutil def throttle_cpu(max_percent50): p psutil.Process() p.cpu_percent() # 首次调用初始化 while True: if p.cpu_percent() max_percent: time.sleep(0.1)17. 扩展应用场景17.1 网站访问监控分析Nginx访问日志NGINX_PATTERNS [ r4\d\d\s\d, # 客户端错误 r5\d\d\s\d # 服务器错误 ] def analyze_nginx(line): if any(re.search(p, line) for p in NGINX_PATTERNS): send_alert(fHTTP错误: {line.split()[8]} {line.split()[6]})17.2 数据库日志监控识别SQL慢查询def analyze_mysql(line): if Query_time in line: time float(re.search(rQuery_time: (\d\.\d), line).group(1)) if time 1.0: # 超过1秒的查询 send_alert(f慢查询: {time}s\n{line})18. 维护与升级策略18.1 配置热更新无需重启加载新配置import signal def reload_config(signum, frame): global config with open(config.yaml) as f: config yaml.safe_load(f) signal.signal(signal.SIGHUP, reload_config)18.2 版本回滚机制使用Git管理配置变更import git repo git.Repo.init(.) def rollback_config(commit_hash): repo.git.checkout(commit_hash, config.yaml) reload_config(None, None)19. 文档与帮助系统19.1 命令行帮助使用argparse实现CLIimport argparse parser argparse.ArgumentParser() parser.add_argument(-c, --config, defaultconfig.yaml) parser.add_argument(-v, --verbose, actionstore_true) args parser.parse_args()19.2 自动生成文档从代码注释生成Markdown文档import inspect def generate_docs(): with open(README.md, w) as f: f.write(# Log Monitor Documentation\n\n) for name, obj in inspect.getmembers(sys.modules[__name__]): if inspect.isfunction(obj): f.write(f## {name}\n\n{inspect.getdoc(obj)}\n\n)20. 项目结构优化推荐的生产级项目布局/logmonitor ├── main.py # 入口脚本 ├── config.yaml # 配置文件 ├── requirements.txt # 依赖列表 ├── src/ │ ├── monitor.py # 监控核心逻辑 │ ├── alert.py # 通知模块 │ └── utils.py # 工具函数 └── tests/ # 测试用例在main.py中实现模块化加载from src.monitor import start_monitoring from src.alert import init_alerts def main(): init_alerts() start_monitoring() if __name__ __main__: main()
Python实现日志监控与警报系统实战指南
1. 项目概述日志监控与警报系统的核心价值服务器日志就像系统的黑匣子记录着每一次请求、每一个错误和每一条关键事件。但海量的日志数据如果不加以监控就如同把金矿埋在地下——直到系统崩溃时才会发现那些早已被记录下来的预警信号。这就是为什么我们需要自动化日志监控系统。这个Python项目解决的核心问题是实时扫描系统日志文件识别预设的关键错误模式如ERROR、CRITICAL等并通过邮件或即时消息平台发送警报通知。相比商业监控工具自建方案的优势在于完全定制化可以针对特定应用定义专属的错误模式零成本利用现有服务器资源无需额外订阅费用深度集成能与内部系统无缝对接不受SaaS产品功能限制2. 技术选型与架构设计2.1 为什么选择PythonPython在日志处理领域有三大不可替代的优势丰富的文本处理库re模块提供强大的正则表达式支持能高效匹配复杂日志模式成熟的邮件发送方案smtplibemail库组合可以处理各种邮件格式和附件跨平台文件监控watchdog库能可靠地检测文件变化兼容Linux/Windows系统2.2 系统架构设计典型的日志监控系统包含以下组件日志文件 → 文件监控 → 模式匹配 → 警报触发 → 通知发送我们采用生产者-消费者模型生产者线程使用watchdog监控日志目录变化消费者线程用正则表达式匹配关键错误通过SMTP发送邮件3. 核心实现步骤详解3.1 环境准备与依赖安装首先确保Python 3.6环境然后安装必要依赖pip install watchdog python-dotenv提示使用python-dotenv管理敏感配置如邮件密码不要硬编码在脚本中3.2 日志监控模块实现使用watchdog的FileSystemEventHandler类创建自定义处理器from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogHandler(FileSystemEventHandler): def on_modified(self, event): if not event.is_directory and event.src_path.endswith(.log): with open(event.src_path) as f: new_lines f.readlines()[-10:] # 读取最后10行新日志 analyze_logs(new_lines)3.3 日志分析引擎核心的正则匹配逻辑示例import re ERROR_PATTERNS [ rERROR.*(timeout|failed), rCRITICAL.*disk\sfull, rAlert:\sCPU\soverload ] def analyze_logs(lines): for line in lines: for pattern in ERROR_PATTERNS: if re.search(pattern, line, re.IGNORECASE): send_alert(f匹配到错误: {pattern}\n日志内容: {line}) break3.4 邮件警报系统使用smtplib发送HTML格式警报邮件import smtplib from email.mime.text import MIMEText from email.header import Header def send_alert(content): msg MIMEText(content, html, utf-8) msg[Subject] Header( 系统异常警报, utf-8) msg[From] monitoryourdomain.com msg[To] adminyourdomain.com with smtplib.SMTP(smtp.server.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)4. 高级功能扩展4.1 频率限制与告警聚合避免短时间内重复发送相同错误的警报from collections import defaultdict from datetime import datetime, timedelta alert_history defaultdict(list) def should_send_alert(error_type): now datetime.now() # 相同错误1小时内不重复发送 recent_alerts [t for t in alert_history[error_type] if now - t timedelta(hours1)] alert_history[error_type].append(now) return len(recent_alerts) 04.2 多通知渠道集成除了邮件还可以添加Slack/webhook支持import requests def send_to_slack(message): webhook_url https://hooks.slack.com/services/XXX payload {text: message} requests.post(webhook_url, jsonpayload)5. 生产环境部署方案5.1 系统服务化部署使用systemd管理监控进程Linux环境# /etc/systemd/system/logmonitor.service [Unit] DescriptionLog Monitor Service [Service] ExecStart/usr/bin/python3 /opt/logmonitor/main.py Restartalways Userroot [Install] WantedBymulti-user.target5.2 日志轮转处理处理logrotate场景的完整方案def handle_rotated_file(file_path): # 检查文件是否被轮转inode变化 current_inode os.stat(file_path).st_ino if current_inode ! getattr(handle_rotated_file, last_inode, None): handle_rotated_file.last_inode current_inode # 重新打开文件读取最新内容 with open(file_path) as f: analyze_logs(f.readlines()[-100:]) # 读取轮转后文件末尾6. 性能优化技巧6.1 高效文件读取方案避免重复读取整个文件def tail(filename, n10): 返回文件最后n行 with open(filename, rb) as f: # 从文件末尾开始读取 f.seek(0, 2) end f.tell() lines_found 0 offset 1 while lines_found n and offset end: f.seek(max(0, end - offset)) data f.read(min(1024, offset)) lines_found data.count(b\n) offset * 2 f.seek(max(0, end - offset)) return f.readlines()[-n:]6.2 正则表达式优化预编译正则模式提升匹配速度COMPILED_PATTERNS [re.compile(p, re.IGNORECASE) for p in ERROR_PATTERNS] def analyze_logs(lines): for line in lines: for pattern in COMPILED_PATTERNS: if pattern.search(line): # ...警报逻辑7. 常见问题与解决方案7.1 文件权限问题典型错误PermissionError: [Errno 13] Permission denied解决方案确保运行用户有日志文件读取权限对于docker容器需要正确挂载volumedocker run -v /var/log:/host_logs logmonitor7.2 字符编码问题处理不同编码的日志文件def try_decode(line): for encoding in [utf-8, gbk, latin-1]: try: return line.decode(encoding) except UnicodeDecodeError: continue return line.decode(utf-8, errorsreplace)7.3 邮件发送失败SMTP常见问题排查检查防火墙是否开放587端口验证是否开启SMTP认证测试Telnet连接telnet smtp.server.com 5878. 监控指标与可视化8.1 关键指标收集记录监控系统自身运行状态monitor_metrics { files_processed: 0, errors_found: 0, last_alert: None } def export_metrics(): return ( f# HELP logmonitor_errors Total errors detected\n f# TYPE logmonitor_errors counter\n flogmonitor_errors {monitor_metrics[errors_found]}\n )8.2 Prometheus集成暴露监控指标端点from http.server import HTTPServer, BaseHTTPRequestHandler class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /metrics: self.send_response(200) self.end_headers() self.wfile.write(export_metrics().encode()) HTTPServer((0.0.0.0, 8000), MetricsHandler).serve_forever()9. 安全最佳实践9.1 敏感信息处理避免日志中的密码泄露SENSITIVE_KEYS [password, secret, api_key] def sanitize_log(line): for key in SENSITIVE_KEYS: line re.sub(fr{key}[^\s], f{key}[REDACTED], line) return line9.2 访问控制限制监控目录范围ALLOWED_PATHS [/var/log/app, /tmp/logs] def is_allowed(path): return any(path.startswith(allowed) for allowed in ALLOWED_PATHS)10. 测试方案设计10.1 单元测试用例测试日志分析核心逻辑import unittest class TestLogAnalysis(unittest.TestCase): def test_error_detection(self): test_lines [ INFO: System started, ERROR: Database connection failed, CRITICAL: Disk space exhausted ] with self.assertLogs() as cm: analyze_logs(test_lines) self.assertIn(ERROR, cm.output[0])10.2 集成测试方案模拟完整工作流程from unittest.mock import patch def test_integration(): with patch(smtplib.SMTP) as mock_smtp: # 触发文件修改事件 event type(, (), {is_directory: False, src_path: test.log})() handler LogHandler() handler.on_modified(event) # 验证邮件发送被调用 assert mock_smtp.return_value.send_message.called11. 性能基准测试使用不同日志量测试处理速度import timeit def benchmark(): setup from __main__ import analyze_logs test_data [ERROR: test] * 10000 stmt analyze_logs(test_data) return timeit.timeit(stmt, setup, number100) print(f处理速度: {benchmark():.2f}秒/万条)12. 容器化部署Dockerfile最佳实践FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [python, main.py]构建与运行docker build -t logmonitor . docker run -d -v /var/log:/logs logmonitor13. 配置管理方案使用YAML文件管理配置# config.yaml monitor: paths: - /var/log/nginx - /opt/app/logs patterns: - ERROR - CRITICAL alert: email: server: smtp.example.com recipients: - adminexample.comPython读取配置import yaml with open(config.yaml) as f: config yaml.safe_load(f)14. 高可用设计14.1 心跳检测机制实现监控进程的健康检查import threading def heart_beat(): while True: with open(/tmp/logmonitor.heartbeat, w) as f: f.write(str(time.time())) time.sleep(60) threading.Thread(targetheart_beat, daemonTrue).start()14.2 分布式监控方案使用Redis实现多节点协同import redis r redis.Redis() def acquire_lock(lock_name, expire300): return r.set(lock_name, 1, nxTrue, exexpire) if acquire_lock(logmonitor:master): # 当前节点获得master角色 start_monitoring()15. 日志分析算法进阶15.1 异常检测算法基于统计的异常值检测import numpy as np def detect_anomaly(error_counts, window24): 使用Z-score检测异常错误量 avg np.mean(error_counts[-window:]) std np.std(error_counts[-window:]) latest error_counts[-1] return abs(latest - avg) 3 * std15.2 机器学习分类使用scikit-learn进行日志分类from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression vectorizer TfidfVectorizer() classifier LogisticRegression() # 训练样本示例 train_texts [ERROR: db fail, INFO: system ok] train_labels [1, 0] X vectorizer.fit_transform(train_texts) classifier.fit(X, train_labels) def classify_log(line): return classifier.predict(vectorizer.transform([line]))16. 系统资源管理16.1 内存限制方案防止日志文件过大导致OOMimport resource def set_memory_limit(percent0.8): soft, hard resource.getrlimit(resource.RLIMIT_AS) total_mem os.sysconf(SC_PAGE_SIZE) * os.sysconf(SC_PHYS_PAGES) new_limit int(total_mem * percent) resource.setrlimit(resource.RLIMIT_AS, (new_limit, hard))16.2 CPU节流控制限制日志分析CPU占用import psutil def throttle_cpu(max_percent50): p psutil.Process() p.cpu_percent() # 首次调用初始化 while True: if p.cpu_percent() max_percent: time.sleep(0.1)17. 扩展应用场景17.1 网站访问监控分析Nginx访问日志NGINX_PATTERNS [ r4\d\d\s\d, # 客户端错误 r5\d\d\s\d # 服务器错误 ] def analyze_nginx(line): if any(re.search(p, line) for p in NGINX_PATTERNS): send_alert(fHTTP错误: {line.split()[8]} {line.split()[6]})17.2 数据库日志监控识别SQL慢查询def analyze_mysql(line): if Query_time in line: time float(re.search(rQuery_time: (\d\.\d), line).group(1)) if time 1.0: # 超过1秒的查询 send_alert(f慢查询: {time}s\n{line})18. 维护与升级策略18.1 配置热更新无需重启加载新配置import signal def reload_config(signum, frame): global config with open(config.yaml) as f: config yaml.safe_load(f) signal.signal(signal.SIGHUP, reload_config)18.2 版本回滚机制使用Git管理配置变更import git repo git.Repo.init(.) def rollback_config(commit_hash): repo.git.checkout(commit_hash, config.yaml) reload_config(None, None)19. 文档与帮助系统19.1 命令行帮助使用argparse实现CLIimport argparse parser argparse.ArgumentParser() parser.add_argument(-c, --config, defaultconfig.yaml) parser.add_argument(-v, --verbose, actionstore_true) args parser.parse_args()19.2 自动生成文档从代码注释生成Markdown文档import inspect def generate_docs(): with open(README.md, w) as f: f.write(# Log Monitor Documentation\n\n) for name, obj in inspect.getmembers(sys.modules[__name__]): if inspect.isfunction(obj): f.write(f## {name}\n\n{inspect.getdoc(obj)}\n\n)20. 项目结构优化推荐的生产级项目布局/logmonitor ├── main.py # 入口脚本 ├── config.yaml # 配置文件 ├── requirements.txt # 依赖列表 ├── src/ │ ├── monitor.py # 监控核心逻辑 │ ├── alert.py # 通知模块 │ └── utils.py # 工具函数 └── tests/ # 测试用例在main.py中实现模块化加载from src.monitor import start_monitoring from src.alert import init_alerts def main(): init_alerts() start_monitoring() if __name__ __main__: main()
监控与优化方案实践)
