在实际运维和开发工作中DNS 服务的稳定性直接关系到整个应用的可用性。手动检查 DNS 解析不仅效率低下也无法及时发现偶发性故障。因此一个能够实时监控、记录历史数据并主动告警的 DNS 监控面板是保障线上服务稳定性的重要工具。本文将记录一个从零开始借助 AI 编程助手构建一个轻量级 DNS 监控面板的完整过程。这个过程并非一帆风顺包含了技术选型、编码实现、环境配置、问题排查以及最终部署上线的全链路实践。无论你是希望了解 DNS 监控的实现原理还是想学习如何高效利用 AI 辅助工具进行项目开发这篇文章都将提供一份可复现的详细指南。1. 理解 DNS 监控的核心需求与技术选型在动手之前我们需要明确这个监控面板究竟要做什么以及如何选择合适的技术栈来实现它。1.1 DNS 监控的核心功能一个基础的 DNS 监控面板至少需要实现以下功能定时探测周期性地向指定的 DNS 服务器发起对特定域名的解析请求。结果记录记录每次探测的结果包括解析出的 IP 地址、响应时间、是否成功等。状态展示通过 Web 界面直观地展示各监控项当前的健康状态和历史趋势。异常告警当解析失败、响应超时或解析结果与预期不符时能通过邮件、Webhook 等方式发出告警。1.2 技术栈选型与考量基于上述需求我们选择以下技术栈并解释其背后的原因后端语言Python原因Python 拥有丰富的网络库如dnspython和 HTTP 框架如Flask,FastAPI非常适合快速开发此类监控工具。其简洁的语法也便于与 AI 助手协作快速迭代代码。Web 框架Flask原因相对于 Django 的“大而全”Flask 更轻量、灵活。对于这个主要提供 API 和简单页面的监控面板来说Flask 足够且学习曲线平缓。数据库SQLite原因项目初期数据量小且希望部署简单无需额外安装数据库服务。SQLite 以单个文件形式存在非常适合作为轻量级数据存储。后期若数据量增大可平滑迁移至 PostgreSQL 或 MySQL。前端HTML JavaScript (Chart.js)原因监控面板的前端交互相对简单主要是表格展示和图表绘制。使用原生技术搭配 Chart.js 图表库可以避免引入复杂的前端框架降低整体复杂度。任务调度APScheduler原因我们需要一个轻量级的库来执行周期性的 DNS 探测任务。APScheduler 功能强大支持多种触发器且易于与 Flask 集成。注意技术选型没有绝对的对错关键在于匹配项目复杂度、团队技能和运维成本。这里的选择侧重于“快速实现”和“易于理解”。2. 项目环境准备与依赖配置在开始编码前我们需要搭建一个干净的开发环境。2.1 创建项目目录与虚拟环境首先在合适的位置创建项目目录并建立 Python 虚拟环境以隔离依赖。# 创建项目目录 mkdir dns-monitor-dashboard cd dns-monitor-dashboard # 创建虚拟环境 (以 Python 3.8 为例) python3 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate激活虚拟环境后命令行提示符通常会发生变化前面会显示(venv)。2.2 安装项目依赖创建一个requirements.txt文件列出所有需要的 Python 包。# requirements.txt Flask2.3.3 dnspython2.4.2 APScheduler3.10.4 Flask-SQLAlchemy3.0.5 Flask-Migrate4.0.5 requests2.31.0然后使用 pip 安装(venv) pip install -r requirements.txt依赖说明Flask: Web 框架核心。dnspython: 用于执行 DNS 查询的专业库。APScheduler: 后台任务调度器。Flask-SQLAlchemy: Flask 的 ORM 扩展用于操作数据库。Flask-Migrate: 配合 SQLAlchemy 进行数据库迁移管理。requests: 用于在告警时发送 HTTP 请求到 Webhook。3. 构建项目结构与核心模型良好的项目结构是代码可维护性的基础。我们采用以下结构dns-monitor-dashboard/ ├── app.py # 应用主入口 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── instance/ # 实例文件夹 (存放数据库文件) ├── migrations/ # 数据库迁移脚本 (由 Flask-Migrate 生成) ├── models.py # 数据模型定义 ├── monitor.py # DNS 监控核心逻辑 ├── scheduler.py # 任务调度器配置 ├── static/ # 静态文件 (CSS, JS) │ └── js/ │ └── chart.js # 可从官网下载 └── templates/ # HTML 模板 └── index.html3.1 定义数据模型 (models.py)我们需要两个核心模型MonitorTarget用于定义监控目标ProbeRecord用于存储每次探测的结果。# models.py from datetime import datetime from app import db class MonitorTarget(db.Model): 监控目标表 id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(80), nullableFalse, uniqueTrue) # 监控项名称 domain db.Column(db.String(255), nullableFalse) # 要解析的域名 dns_server db.Column(db.String(50), nullableFalse) # DNS 服务器地址如 “8.8.8.8” record_type db.Column(db.String(10), defaultA) # 记录类型A, AAAA, CNAME 等 expected_ip db.Column(db.String(255)) # 期望的IP可选用于校验 check_interval db.Column(db.Integer, default60) # 检查间隔秒 is_active db.Column(db.Boolean, defaultTrue) # 是否启用监控 created_at db.Column(db.DateTime, defaultdatetime.utcnow) # 定义关系一个监控目标有多条探测记录 records db.relationship(ProbeRecord, backreftarget, lazyTrue, cascadeall, delete-orphan) def __repr__(self): return fMonitorTarget {self.name} class ProbeRecord(db.Model): 探测记录表 id db.Column(db.Integer, primary_keyTrue) target_id db.Column(db.Integer, db.ForeignKey(monitor_target.id), nullableFalse) resolved_ips db.Column(db.Text) # 解析出的IP列表用逗号分隔 response_time db.Column(db.Float) # 响应时间毫秒 status db.Column(db.String(20)) # 状态success, timeout, error error_message db.Column(db.Text) # 错误信息如果有 created_at db.Column(db.DateTime, defaultdatetime.utcnow, indexTrue) # 加索引便于按时间查询 def __repr__(self): return fProbeRecord {self.target_id} {self.status}关键点解释db.relationship: 定义了MonitorTarget和ProbeRecord之间的一对多关系方便通过target.records查询某个目标的所有历史记录。cascadeall, delete-orphan: 设置级联删除当删除一个监控目标时其关联的所有探测记录也会被自动删除。indexTrue: 为created_at字段创建索引能大幅提升按时间范围查询历史记录的性能。3.2 应用配置与初始化 (app.py, config.py)创建 Flask 应用实例并加载配置。# config.py import os basedir os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or dev-secret-key-change-in-production SQLALCHEMY_DATABASE_URI os.environ.get(DATABASE_URL) or \ sqlite:/// os.path.join(basedir, instance, monitor.db) SQLALCHEMY_TRACK_MODIFICATIONS False SCHEDULER_API_ENABLED False # 禁用 APScheduler 的 REST API安全考虑# app.py from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from config import Config # 初始化扩展 db SQLAlchemy() migrate Migrate() def create_app(config_classConfig): app Flask(__name__) app.config.from_object(config_class) # 初始化扩展 db.init_app(app) migrate.init_app(app, db) # 在这里注册蓝图后续添加 # from app.routes import bp # app.register_blueprint(bp) # 确保 instance 文件夹存在 try: os.makedirs(app.instance_path) except OSError: pass return app # 创建应用实例 app create_app() # 导入模型确保它们被加载并与 db 实例关联 from models import MonitorTarget, ProbeRecord4. 实现 DNS 探测与任务调度逻辑这是整个项目的核心负责执行 DNS 查询并将结果存入数据库。4.1 编写 DNS 探测函数 (monitor.py)# monitor.py import dns.resolver import dns.exception from datetime import datetime from app import db from models import MonitorTarget, ProbeRecord def probe_dns_target(target): 对单个监控目标执行 DNS 探测。 :param target: MonitorTarget 实例 :return: ProbeRecord 实例已创建但未提交到数据库 resolver dns.resolver.Resolver() resolver.nameservers [target.dns_server] # 指定 DNS 服务器 resolver.timeout 5.0 # 查询超时时间秒 resolver.lifetime 5.0 # 总超时时间秒 record ProbeRecord(target_idtarget.id) start_time datetime.now() try: answers resolver.resolve(target.domain, target.record_type) end_time datetime.now() # 计算响应时间毫秒 response_time_ms (end_time - start_time).total_seconds() * 1000 record.response_time round(response_time_ms, 2) # 收集解析出的 IP 地址 ips [rdata.to_text() for rdata in answers] record.resolved_ips , .join(ips) record.status success # 可选校验预期 IP if target.expected_ip: expected_set set(target.expected_ip.split(,)) resolved_set set(ips) if not resolved_set.issuperset(expected_set): record.status mismatch record.error_message fResolved {record.resolved_ips}, expected {target.expected_ip} except dns.resolver.Timeout: record.status timeout record.error_message DNS query timed out. except dns.exception.DNSException as e: record.status error record.error_message str(e) except Exception as e: record.status error record.error_message fUnexpected error: {str(e)} return record关键参数与错误处理resolver.timeout和resolver.lifetime控制单次查询和总查询的超时防止因网络或 DNS 服务器问题导致任务长时间阻塞。异常捕获dns.resolver.Timeout捕获超时dns.exception.DNSException捕获 DNS 相关错误如 NXDOMAIN最外层的Exception捕获其他未知异常。清晰的错误分类便于后续告警和排查。4.2 配置后台调度任务 (scheduler.py)我们需要一个函数来遍历所有活跃的监控目标并执行探测然后让 APScheduler 周期性地调用这个函数。# scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from app import create_app, db from models import MonitorTarget from monitor import probe_dns_target import atexit import logging logging.basicConfig() logging.getLogger(apscheduler).setLevel(logging.WARNING) def probe_all_targets(): 探测所有活跃的监控目标 app create_app() with app.app_context(): active_targets MonitorTarget.query.filter_by(is_activeTrue).all() for target in active_targets: print(fProbing {target.name} ({target.domain}) via {target.dns_server}...) try: record probe_dns_target(target) db.session.add(record) db.session.commit() print(f - Status: {record.status}, Time: {record.response_time}ms) except Exception as e: db.session.rollback() print(f - Failed to save record for {target.name}: {e}) def init_scheduler(): 初始化并启动调度器 scheduler BackgroundScheduler() # 每30秒执行一次探测任务。实际项目中应根据监控目标数量和间隔动态调整。 trigger IntervalTrigger(seconds30) scheduler.add_job(funcprobe_all_targets, triggertrigger, idprobe_job, max_instances1) scheduler.start() print(Scheduler started.) # 优雅关闭 atexit.register(lambda: scheduler.shutdown())调度策略说明IntervalTrigger(seconds30)每 30 秒触发一次。这个间隔是全局的所有目标的探测都会在这个周期内执行。如果目标很多或间隔很短可能导致单次执行时间过长。更精细的控制可以为每个MonitorTarget设置独立的调度任务但复杂度会显著增加。max_instances1确保同一时间只有一个探测任务在执行避免并发写入数据库导致冲突。app.app_context()因为探测和数据库操作需要 Flask 应用上下文我们在任务函数内部创建了它。5. 创建 Web 界面与 API现在我们需要一个界面来展示监控状态和历史数据。5.1 设计前端页面 (templates/index.html)这是一个简单的单页应用使用 Chart.js 绘制响应时间趋势图。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleDNS 监控面板/title script srchttps://cdn.jsdelivr.net/npm/chart.js/script style body { font-family: sans-serif; margin: 20px; } .status { padding: 5px 10px; border-radius: 3px; color: white; } .success { background-color: #28a745; } .error, .timeout, .mismatch { background-color: #dc3545; } table { border-collapse: collapse; width: 100%; margin-top: 20px; } th, td { border: 1px solid #ddd; padding: 8px; text-align: left; } th { background-color: #f2f2f2; } .chart-container { margin-top: 40px; width: 80%; height: 400px; } /style /head body h1DNS 监控面板/h1 div idtargets-summary p正在加载监控目标状态.../p /div div classchart-container canvas idresponseTimeChart/canvas /div script let myChart; async function fetchData() { try { const response await fetch(/api/targets/status); const data await response.json(); renderSummary(data); updateChart(data); } catch (error) { console.error(获取数据失败:, error); document.getElementById(targets-summary).innerHTML p stylecolor:red;加载数据失败: ${error.message}/p; } } function renderSummary(data) { let html table trth名称/thth域名/ththDNS 服务器/thth最后状态/thth最后响应时间/thth最后检查时间/th/tr; data.forEach(target { const lastRecord target.last_record; const statusClass (lastRecord lastRecord.status success) ? success : error; const statusText lastRecord ? lastRecord.status : N/A; const responseTime lastRecord ? ${lastRecord.response_time} ms : N/A; const checkTime lastRecord ? new Date(lastRecord.created_at).toLocaleString() : N/A; html tr td${target.name}/td td${target.domain}/td td${target.dns_server}/td tdspan classstatus ${statusClass}${statusText}/span/td td${responseTime}/td td${checkTime}/td /tr; }); html /table; document.getElementById(targets-summary).innerHTML html; } function updateChart(data) { const ctx document.getElementById(responseTimeChart).getContext(2d); const labels data.map(t t.name); const lastResponseTimes data.map(t t.last_record ? t.last_record.response_time : 0); if (myChart) { myChart.destroy(); } myChart new Chart(ctx, { type: bar, data: { labels: labels, datasets: [{ label: 最近一次响应时间 (ms), data: lastResponseTimes, backgroundColor: lastResponseTimes.map(time time 100 ? rgba(220, 53, 69, 0.7) : rgba(40, 167, 69, 0.7)), borderColor: rgba(54, 162, 235, 1), borderWidth: 1 }] }, options: { scales: { y: { beginAtZero: true, title: { display: true, text: 响应时间 (ms) } } } } }); } // 页面加载时获取数据并每30秒刷新一次 fetchData(); setInterval(fetchData, 30000); /script /body /html5.2 提供后端 API (routes.py)创建路由文件为前端提供数据接口。# routes.py from flask import Blueprint, render_template, jsonify from app import db from models import MonitorTarget, ProbeRecord from sqlalchemy import desc bp Blueprint(main, __name__) bp.route(/) def index(): 渲染监控面板首页 return render_template(index.html) bp.route(/api/targets/status) def get_targets_status(): 获取所有监控目标及其最新状态 targets MonitorTarget.query.filter_by(is_activeTrue).all() result [] for target in targets: # 获取该目标最近的一条探测记录 last_record ProbeRecord.query.filter_by(target_idtarget.id).order_by(desc(ProbeRecord.created_at)).first() target_data { id: target.id, name: target.name, domain: target.domain, dns_server: target.dns_server, check_interval: target.check_interval, last_record: None } if last_record: target_data[last_record] { status: last_record.status, response_time: last_record.response_time, resolved_ips: last_record.resolved_ips, created_at: last_record.created_at.isoformat() } result.append(target_data) return jsonify(result)然后在app.py中注册这个蓝图# app.py (在 create_app 函数内添加) def create_app(config_classConfig): # ... 之前的代码 ... from routes import bp app.register_blueprint(bp) # ... 之后的代码 ...6. 初始化、运行与验证所有代码就绪后我们需要初始化数据库并启动应用。6.1 初始化数据库首先设置 Flask 应用上下文并创建数据库迁移目录如果尚未创建。(venv) export FLASK_APPapp.py # Linux/macOS # (venv) set FLASK_APPapp.py # Windows (venv) flask db init (venv) flask db migrate -m Initial migration. (venv) flask db upgrade执行成功后会在instance文件夹下生成monitor.db数据库文件。6.2 添加测试监控目标我们可以写一个简单的脚本或者直接使用 Flask Shell 来添加几个监控目标。(venv) flask shell在打开的 Python Shell 中执行 from app import db from models import MonitorTarget target1 MonitorTarget(nameGoogle DNS, domaingoogle.com, dns_server8.8.8.8, check_interval60) target2 MonitorTarget(nameCloudflare DNS, domaincloudflare.com, dns_server1.1.1.1, check_interval60) target3 MonitorTarget(nameLocal Check, domainbaidu.com, dns_server114.114.114.114, check_interval30, expected_ip39.156.66.10,110.242.68.66) db.session.add_all([target1, target2, target3]) db.session.commit() exit()6.3 启动应用与调度器这是关键一步需要同时启动 Flask Web 服务器和后台调度器。我们修改app.py的启动部分。# app.py (文件末尾) if __name__ __main__: # 注意这种启动方式仅适用于开发环境。 # 在生产环境中应使用 Gunicorn/uWSGI 单独的进程管理调度器。 from scheduler import init_scheduler init_scheduler() # 启动后台调度任务 app.run(debugTrue, host0.0.0.0, port5000, use_reloaderFalse) # 禁用重载器避免调度器被重复启动重要提示use_reloaderFalse是必须的。Flask 的调试重载器会重新加载应用导致 APScheduler 被重复初始化并启动多个实例。现在在项目根目录下运行(venv) python app.py你应该能看到类似以下的输出Scheduler started. * Serving Flask app app * Debug mode: on WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Running on all addresses (0.0.0.0) * Running on http://127.0.0.1:5000 * Running on http://192.168.x.x:5000 Press CTRLC to quit6.4 验证功能访问面板打开浏览器访问http://127.0.0.1:5000。你应该能看到一个表格显示刚刚添加的三个监控目标。初始状态可能为“N/A”因为调度器需要运行一个周期30秒后才有数据。观察日志在运行app.py的控制台每隔30秒会打印出探测任务的日志显示每个目标的探测状态和响应时间。检查数据库等待几分钟后刷新页面表格中会显示最新的状态、响应时间和检查时间。图表也会根据响应时间更新。模拟故障你可以尝试修改一个监控目标的 DNS 服务器地址为一个不可达的 IP如192.0.2.1观察其状态是否会变为timeout或error。7. 常见问题排查与优化在实际开发和部署中你可能会遇到以下问题。7.1 问题排查清单问题现象可能原因检查方式处理建议应用启动后无探测日志调度器未成功启动1. 检查app.py末尾是否调用了init_scheduler()。2. 检查use_reloaderFalse是否设置。3. 查看启动日志是否有Scheduler started.。确保调度器初始化代码在if __name__ __main__:块内且重载器已禁用。数据库无记录或记录异常数据库连接问题或模型定义错误1. 检查instance/monitor.db文件是否存在。2. 在 Flask Shell 中手动执行probe_all_targets函数看是否报错。3. 检查models.py中字段名是否与查询代码一致。使用flask shell进行交互式调试。确认db.session.commit()被成功执行。前端页面无法加载或图表不显示静态文件路径错误或 API 返回异常1. 打开浏览器开发者工具查看 Console 和 Network 标签页。2. 直接访问http://127.0.0.1:5000/api/targets/status看是否返回 JSON 数据。3. 检查templates/index.html中 Chart.js 的 CDN 链接是否有效。根据浏览器报错信息定位问题。确保后端 API 路由正确定义并注册。探测任务执行时间过长影响 Web 响应监控目标过多或 DNS 服务器响应慢1. 查看控制台日志计算probe_all_targets函数的执行时间。2. 检查是否有目标配置了响应很慢的 DNS 服务器。1. 增加resolver.timeout和lifetime避免单个查询阻塞太久。2. 考虑使用线程池并发执行探测需注意数据库会话管理。3. 调整全局探测间隔。生产环境部署后调度器停止使用开发服务器部署进程不稳定开发服务器 (app.run) 不适合生产环境进程崩溃或重启会导致调度器停止。生产环境应使用Gunicorn或uWSGI运行 Flask并使用Supervisor或systemd单独管理一个常驻的调度器进程。7.2 生产环境部署建议开发环境运行顺利不代表生产环境就高枕无忧。以下是为生产部署必须考虑的几点分离调度器与 Web 服务问题在app.py中启动调度器的方式使得调度器和 Web 服务耦合在同一个进程。当使用多 worker 模式如 Gunicorn部署时每个 worker 都会启动自己的调度器导致任务重复执行。方案将scheduler.py改造成一个独立的脚本如run_scheduler.py使用python run_scheduler.py单独运行。然后使用进程管理工具如 Supervisor来同时管理 Flask 应用进程和调度器进程。配置外置化将数据库连接字符串、密钥、告警 Webhook 地址等敏感信息从代码中移除改为从环境变量或配置文件如.env中读取。可以使用python-dotenv库。添加告警功能在monitor.py的probe_dns_target函数中当record.status不是success时触发告警逻辑如发送邮件、调用钉钉/企业微信机器人 Webhook。注意要添加告警抑制机制避免短时间内重复告警。数据库优化ProbeRecord表会随时间快速增长。需要定期归档或清理旧数据。可以在该表上按created_at字段建立分区或者定期执行清理任务。日志记录将print语句替换为标准的日志记录如 Pythonlogging模块并配置日志级别、格式和输出位置文件、syslog等便于问题追踪。7.3 利用 AI 编程助手的高效协作心得在开发过程中AI 助手能极大提升效率但也可能“翻车”。以下是一些实践建议明确需求在向 AI 提问前自己先理清功能边界、输入输出和异常情况。模糊的指令会得到模糊甚至错误的代码。分步验证不要一次性让 AI 生成全部代码。应该按模块如模型定义、探测函数、API 路由分别生成和测试。每完成一个可运行的小单元就验证一次。理解代码AI 生成的代码一定要读明白再使用。特别是涉及数据库操作、网络请求、并发处理的部分要清楚其原理和潜在风险。处理“幻觉”AI 可能会使用不存在的库函数或错误的参数。遇到报错时要优先查阅官方文档进行核实和修正。迭代优化第一版代码通常只解决“有无”问题。在跑通基础功能后可以再让 AI 协助进行代码重构、添加错误处理、编写单元测试等。通过这个项目我们不仅实现了一个可用的 DNS 监控面板更实践了一条从需求分析、技术选型、编码实现、问题排查到生产准备的完整开发路径。你可以在此基础上继续扩展告警、多用户、更丰富的图表分析等功能使其更贴合你的实际运维场景。 从零构建轻量级DNS监控面板：Python+Flask+AI编程实践
在实际运维和开发工作中DNS 服务的稳定性直接关系到整个应用的可用性。手动检查 DNS 解析不仅效率低下也无法及时发现偶发性故障。因此一个能够实时监控、记录历史数据并主动告警的 DNS 监控面板是保障线上服务稳定性的重要工具。本文将记录一个从零开始借助 AI 编程助手构建一个轻量级 DNS 监控面板的完整过程。这个过程并非一帆风顺包含了技术选型、编码实现、环境配置、问题排查以及最终部署上线的全链路实践。无论你是希望了解 DNS 监控的实现原理还是想学习如何高效利用 AI 辅助工具进行项目开发这篇文章都将提供一份可复现的详细指南。1. 理解 DNS 监控的核心需求与技术选型在动手之前我们需要明确这个监控面板究竟要做什么以及如何选择合适的技术栈来实现它。1.1 DNS 监控的核心功能一个基础的 DNS 监控面板至少需要实现以下功能定时探测周期性地向指定的 DNS 服务器发起对特定域名的解析请求。结果记录记录每次探测的结果包括解析出的 IP 地址、响应时间、是否成功等。状态展示通过 Web 界面直观地展示各监控项当前的健康状态和历史趋势。异常告警当解析失败、响应超时或解析结果与预期不符时能通过邮件、Webhook 等方式发出告警。1.2 技术栈选型与考量基于上述需求我们选择以下技术栈并解释其背后的原因后端语言Python原因Python 拥有丰富的网络库如dnspython和 HTTP 框架如Flask,FastAPI非常适合快速开发此类监控工具。其简洁的语法也便于与 AI 助手协作快速迭代代码。Web 框架Flask原因相对于 Django 的“大而全”Flask 更轻量、灵活。对于这个主要提供 API 和简单页面的监控面板来说Flask 足够且学习曲线平缓。数据库SQLite原因项目初期数据量小且希望部署简单无需额外安装数据库服务。SQLite 以单个文件形式存在非常适合作为轻量级数据存储。后期若数据量增大可平滑迁移至 PostgreSQL 或 MySQL。前端HTML JavaScript (Chart.js)原因监控面板的前端交互相对简单主要是表格展示和图表绘制。使用原生技术搭配 Chart.js 图表库可以避免引入复杂的前端框架降低整体复杂度。任务调度APScheduler原因我们需要一个轻量级的库来执行周期性的 DNS 探测任务。APScheduler 功能强大支持多种触发器且易于与 Flask 集成。注意技术选型没有绝对的对错关键在于匹配项目复杂度、团队技能和运维成本。这里的选择侧重于“快速实现”和“易于理解”。2. 项目环境准备与依赖配置在开始编码前我们需要搭建一个干净的开发环境。2.1 创建项目目录与虚拟环境首先在合适的位置创建项目目录并建立 Python 虚拟环境以隔离依赖。# 创建项目目录 mkdir dns-monitor-dashboard cd dns-monitor-dashboard # 创建虚拟环境 (以 Python 3.8 为例) python3 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate激活虚拟环境后命令行提示符通常会发生变化前面会显示(venv)。2.2 安装项目依赖创建一个requirements.txt文件列出所有需要的 Python 包。# requirements.txt Flask2.3.3 dnspython2.4.2 APScheduler3.10.4 Flask-SQLAlchemy3.0.5 Flask-Migrate4.0.5 requests2.31.0然后使用 pip 安装(venv) pip install -r requirements.txt依赖说明Flask: Web 框架核心。dnspython: 用于执行 DNS 查询的专业库。APScheduler: 后台任务调度器。Flask-SQLAlchemy: Flask 的 ORM 扩展用于操作数据库。Flask-Migrate: 配合 SQLAlchemy 进行数据库迁移管理。requests: 用于在告警时发送 HTTP 请求到 Webhook。3. 构建项目结构与核心模型良好的项目结构是代码可维护性的基础。我们采用以下结构dns-monitor-dashboard/ ├── app.py # 应用主入口 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── instance/ # 实例文件夹 (存放数据库文件) ├── migrations/ # 数据库迁移脚本 (由 Flask-Migrate 生成) ├── models.py # 数据模型定义 ├── monitor.py # DNS 监控核心逻辑 ├── scheduler.py # 任务调度器配置 ├── static/ # 静态文件 (CSS, JS) │ └── js/ │ └── chart.js # 可从官网下载 └── templates/ # HTML 模板 └── index.html3.1 定义数据模型 (models.py)我们需要两个核心模型MonitorTarget用于定义监控目标ProbeRecord用于存储每次探测的结果。# models.py from datetime import datetime from app import db class MonitorTarget(db.Model): 监控目标表 id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(80), nullableFalse, uniqueTrue) # 监控项名称 domain db.Column(db.String(255), nullableFalse) # 要解析的域名 dns_server db.Column(db.String(50), nullableFalse) # DNS 服务器地址如 “8.8.8.8” record_type db.Column(db.String(10), defaultA) # 记录类型A, AAAA, CNAME 等 expected_ip db.Column(db.String(255)) # 期望的IP可选用于校验 check_interval db.Column(db.Integer, default60) # 检查间隔秒 is_active db.Column(db.Boolean, defaultTrue) # 是否启用监控 created_at db.Column(db.DateTime, defaultdatetime.utcnow) # 定义关系一个监控目标有多条探测记录 records db.relationship(ProbeRecord, backreftarget, lazyTrue, cascadeall, delete-orphan) def __repr__(self): return fMonitorTarget {self.name} class ProbeRecord(db.Model): 探测记录表 id db.Column(db.Integer, primary_keyTrue) target_id db.Column(db.Integer, db.ForeignKey(monitor_target.id), nullableFalse) resolved_ips db.Column(db.Text) # 解析出的IP列表用逗号分隔 response_time db.Column(db.Float) # 响应时间毫秒 status db.Column(db.String(20)) # 状态success, timeout, error error_message db.Column(db.Text) # 错误信息如果有 created_at db.Column(db.DateTime, defaultdatetime.utcnow, indexTrue) # 加索引便于按时间查询 def __repr__(self): return fProbeRecord {self.target_id} {self.status}关键点解释db.relationship: 定义了MonitorTarget和ProbeRecord之间的一对多关系方便通过target.records查询某个目标的所有历史记录。cascadeall, delete-orphan: 设置级联删除当删除一个监控目标时其关联的所有探测记录也会被自动删除。indexTrue: 为created_at字段创建索引能大幅提升按时间范围查询历史记录的性能。3.2 应用配置与初始化 (app.py, config.py)创建 Flask 应用实例并加载配置。# config.py import os basedir os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or dev-secret-key-change-in-production SQLALCHEMY_DATABASE_URI os.environ.get(DATABASE_URL) or \ sqlite:/// os.path.join(basedir, instance, monitor.db) SQLALCHEMY_TRACK_MODIFICATIONS False SCHEDULER_API_ENABLED False # 禁用 APScheduler 的 REST API安全考虑# app.py from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from config import Config # 初始化扩展 db SQLAlchemy() migrate Migrate() def create_app(config_classConfig): app Flask(__name__) app.config.from_object(config_class) # 初始化扩展 db.init_app(app) migrate.init_app(app, db) # 在这里注册蓝图后续添加 # from app.routes import bp # app.register_blueprint(bp) # 确保 instance 文件夹存在 try: os.makedirs(app.instance_path) except OSError: pass return app # 创建应用实例 app create_app() # 导入模型确保它们被加载并与 db 实例关联 from models import MonitorTarget, ProbeRecord4. 实现 DNS 探测与任务调度逻辑这是整个项目的核心负责执行 DNS 查询并将结果存入数据库。4.1 编写 DNS 探测函数 (monitor.py)# monitor.py import dns.resolver import dns.exception from datetime import datetime from app import db from models import MonitorTarget, ProbeRecord def probe_dns_target(target): 对单个监控目标执行 DNS 探测。 :param target: MonitorTarget 实例 :return: ProbeRecord 实例已创建但未提交到数据库 resolver dns.resolver.Resolver() resolver.nameservers [target.dns_server] # 指定 DNS 服务器 resolver.timeout 5.0 # 查询超时时间秒 resolver.lifetime 5.0 # 总超时时间秒 record ProbeRecord(target_idtarget.id) start_time datetime.now() try: answers resolver.resolve(target.domain, target.record_type) end_time datetime.now() # 计算响应时间毫秒 response_time_ms (end_time - start_time).total_seconds() * 1000 record.response_time round(response_time_ms, 2) # 收集解析出的 IP 地址 ips [rdata.to_text() for rdata in answers] record.resolved_ips , .join(ips) record.status success # 可选校验预期 IP if target.expected_ip: expected_set set(target.expected_ip.split(,)) resolved_set set(ips) if not resolved_set.issuperset(expected_set): record.status mismatch record.error_message fResolved {record.resolved_ips}, expected {target.expected_ip} except dns.resolver.Timeout: record.status timeout record.error_message DNS query timed out. except dns.exception.DNSException as e: record.status error record.error_message str(e) except Exception as e: record.status error record.error_message fUnexpected error: {str(e)} return record关键参数与错误处理resolver.timeout和resolver.lifetime控制单次查询和总查询的超时防止因网络或 DNS 服务器问题导致任务长时间阻塞。异常捕获dns.resolver.Timeout捕获超时dns.exception.DNSException捕获 DNS 相关错误如 NXDOMAIN最外层的Exception捕获其他未知异常。清晰的错误分类便于后续告警和排查。4.2 配置后台调度任务 (scheduler.py)我们需要一个函数来遍历所有活跃的监控目标并执行探测然后让 APScheduler 周期性地调用这个函数。# scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from app import create_app, db from models import MonitorTarget from monitor import probe_dns_target import atexit import logging logging.basicConfig() logging.getLogger(apscheduler).setLevel(logging.WARNING) def probe_all_targets(): 探测所有活跃的监控目标 app create_app() with app.app_context(): active_targets MonitorTarget.query.filter_by(is_activeTrue).all() for target in active_targets: print(fProbing {target.name} ({target.domain}) via {target.dns_server}...) try: record probe_dns_target(target) db.session.add(record) db.session.commit() print(f - Status: {record.status}, Time: {record.response_time}ms) except Exception as e: db.session.rollback() print(f - Failed to save record for {target.name}: {e}) def init_scheduler(): 初始化并启动调度器 scheduler BackgroundScheduler() # 每30秒执行一次探测任务。实际项目中应根据监控目标数量和间隔动态调整。 trigger IntervalTrigger(seconds30) scheduler.add_job(funcprobe_all_targets, triggertrigger, idprobe_job, max_instances1) scheduler.start() print(Scheduler started.) # 优雅关闭 atexit.register(lambda: scheduler.shutdown())调度策略说明IntervalTrigger(seconds30)每 30 秒触发一次。这个间隔是全局的所有目标的探测都会在这个周期内执行。如果目标很多或间隔很短可能导致单次执行时间过长。更精细的控制可以为每个MonitorTarget设置独立的调度任务但复杂度会显著增加。max_instances1确保同一时间只有一个探测任务在执行避免并发写入数据库导致冲突。app.app_context()因为探测和数据库操作需要 Flask 应用上下文我们在任务函数内部创建了它。5. 创建 Web 界面与 API现在我们需要一个界面来展示监控状态和历史数据。5.1 设计前端页面 (templates/index.html)这是一个简单的单页应用使用 Chart.js 绘制响应时间趋势图。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleDNS 监控面板/title script srchttps://cdn.jsdelivr.net/npm/chart.js/script style body { font-family: sans-serif; margin: 20px; } .status { padding: 5px 10px; border-radius: 3px; color: white; } .success { background-color: #28a745; } .error, .timeout, .mismatch { background-color: #dc3545; } table { border-collapse: collapse; width: 100%; margin-top: 20px; } th, td { border: 1px solid #ddd; padding: 8px; text-align: left; } th { background-color: #f2f2f2; } .chart-container { margin-top: 40px; width: 80%; height: 400px; } /style /head body h1DNS 监控面板/h1 div idtargets-summary p正在加载监控目标状态.../p /div div classchart-container canvas idresponseTimeChart/canvas /div script let myChart; async function fetchData() { try { const response await fetch(/api/targets/status); const data await response.json(); renderSummary(data); updateChart(data); } catch (error) { console.error(获取数据失败:, error); document.getElementById(targets-summary).innerHTML p stylecolor:red;加载数据失败: ${error.message}/p; } } function renderSummary(data) { let html table trth名称/thth域名/ththDNS 服务器/thth最后状态/thth最后响应时间/thth最后检查时间/th/tr; data.forEach(target { const lastRecord target.last_record; const statusClass (lastRecord lastRecord.status success) ? success : error; const statusText lastRecord ? lastRecord.status : N/A; const responseTime lastRecord ? ${lastRecord.response_time} ms : N/A; const checkTime lastRecord ? new Date(lastRecord.created_at).toLocaleString() : N/A; html tr td${target.name}/td td${target.domain}/td td${target.dns_server}/td tdspan classstatus ${statusClass}${statusText}/span/td td${responseTime}/td td${checkTime}/td /tr; }); html /table; document.getElementById(targets-summary).innerHTML html; } function updateChart(data) { const ctx document.getElementById(responseTimeChart).getContext(2d); const labels data.map(t t.name); const lastResponseTimes data.map(t t.last_record ? t.last_record.response_time : 0); if (myChart) { myChart.destroy(); } myChart new Chart(ctx, { type: bar, data: { labels: labels, datasets: [{ label: 最近一次响应时间 (ms), data: lastResponseTimes, backgroundColor: lastResponseTimes.map(time time 100 ? rgba(220, 53, 69, 0.7) : rgba(40, 167, 69, 0.7)), borderColor: rgba(54, 162, 235, 1), borderWidth: 1 }] }, options: { scales: { y: { beginAtZero: true, title: { display: true, text: 响应时间 (ms) } } } } }); } // 页面加载时获取数据并每30秒刷新一次 fetchData(); setInterval(fetchData, 30000); /script /body /html5.2 提供后端 API (routes.py)创建路由文件为前端提供数据接口。# routes.py from flask import Blueprint, render_template, jsonify from app import db from models import MonitorTarget, ProbeRecord from sqlalchemy import desc bp Blueprint(main, __name__) bp.route(/) def index(): 渲染监控面板首页 return render_template(index.html) bp.route(/api/targets/status) def get_targets_status(): 获取所有监控目标及其最新状态 targets MonitorTarget.query.filter_by(is_activeTrue).all() result [] for target in targets: # 获取该目标最近的一条探测记录 last_record ProbeRecord.query.filter_by(target_idtarget.id).order_by(desc(ProbeRecord.created_at)).first() target_data { id: target.id, name: target.name, domain: target.domain, dns_server: target.dns_server, check_interval: target.check_interval, last_record: None } if last_record: target_data[last_record] { status: last_record.status, response_time: last_record.response_time, resolved_ips: last_record.resolved_ips, created_at: last_record.created_at.isoformat() } result.append(target_data) return jsonify(result)然后在app.py中注册这个蓝图# app.py (在 create_app 函数内添加) def create_app(config_classConfig): # ... 之前的代码 ... from routes import bp app.register_blueprint(bp) # ... 之后的代码 ...6. 初始化、运行与验证所有代码就绪后我们需要初始化数据库并启动应用。6.1 初始化数据库首先设置 Flask 应用上下文并创建数据库迁移目录如果尚未创建。(venv) export FLASK_APPapp.py # Linux/macOS # (venv) set FLASK_APPapp.py # Windows (venv) flask db init (venv) flask db migrate -m Initial migration. (venv) flask db upgrade执行成功后会在instance文件夹下生成monitor.db数据库文件。6.2 添加测试监控目标我们可以写一个简单的脚本或者直接使用 Flask Shell 来添加几个监控目标。(venv) flask shell在打开的 Python Shell 中执行 from app import db from models import MonitorTarget target1 MonitorTarget(nameGoogle DNS, domaingoogle.com, dns_server8.8.8.8, check_interval60) target2 MonitorTarget(nameCloudflare DNS, domaincloudflare.com, dns_server1.1.1.1, check_interval60) target3 MonitorTarget(nameLocal Check, domainbaidu.com, dns_server114.114.114.114, check_interval30, expected_ip39.156.66.10,110.242.68.66) db.session.add_all([target1, target2, target3]) db.session.commit() exit()6.3 启动应用与调度器这是关键一步需要同时启动 Flask Web 服务器和后台调度器。我们修改app.py的启动部分。# app.py (文件末尾) if __name__ __main__: # 注意这种启动方式仅适用于开发环境。 # 在生产环境中应使用 Gunicorn/uWSGI 单独的进程管理调度器。 from scheduler import init_scheduler init_scheduler() # 启动后台调度任务 app.run(debugTrue, host0.0.0.0, port5000, use_reloaderFalse) # 禁用重载器避免调度器被重复启动重要提示use_reloaderFalse是必须的。Flask 的调试重载器会重新加载应用导致 APScheduler 被重复初始化并启动多个实例。现在在项目根目录下运行(venv) python app.py你应该能看到类似以下的输出Scheduler started. * Serving Flask app app * Debug mode: on WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Running on all addresses (0.0.0.0) * Running on http://127.0.0.1:5000 * Running on http://192.168.x.x:5000 Press CTRLC to quit6.4 验证功能访问面板打开浏览器访问http://127.0.0.1:5000。你应该能看到一个表格显示刚刚添加的三个监控目标。初始状态可能为“N/A”因为调度器需要运行一个周期30秒后才有数据。观察日志在运行app.py的控制台每隔30秒会打印出探测任务的日志显示每个目标的探测状态和响应时间。检查数据库等待几分钟后刷新页面表格中会显示最新的状态、响应时间和检查时间。图表也会根据响应时间更新。模拟故障你可以尝试修改一个监控目标的 DNS 服务器地址为一个不可达的 IP如192.0.2.1观察其状态是否会变为timeout或error。7. 常见问题排查与优化在实际开发和部署中你可能会遇到以下问题。7.1 问题排查清单问题现象可能原因检查方式处理建议应用启动后无探测日志调度器未成功启动1. 检查app.py末尾是否调用了init_scheduler()。2. 检查use_reloaderFalse是否设置。3. 查看启动日志是否有Scheduler started.。确保调度器初始化代码在if __name__ __main__:块内且重载器已禁用。数据库无记录或记录异常数据库连接问题或模型定义错误1. 检查instance/monitor.db文件是否存在。2. 在 Flask Shell 中手动执行probe_all_targets函数看是否报错。3. 检查models.py中字段名是否与查询代码一致。使用flask shell进行交互式调试。确认db.session.commit()被成功执行。前端页面无法加载或图表不显示静态文件路径错误或 API 返回异常1. 打开浏览器开发者工具查看 Console 和 Network 标签页。2. 直接访问http://127.0.0.1:5000/api/targets/status看是否返回 JSON 数据。3. 检查templates/index.html中 Chart.js 的 CDN 链接是否有效。根据浏览器报错信息定位问题。确保后端 API 路由正确定义并注册。探测任务执行时间过长影响 Web 响应监控目标过多或 DNS 服务器响应慢1. 查看控制台日志计算probe_all_targets函数的执行时间。2. 检查是否有目标配置了响应很慢的 DNS 服务器。1. 增加resolver.timeout和lifetime避免单个查询阻塞太久。2. 考虑使用线程池并发执行探测需注意数据库会话管理。3. 调整全局探测间隔。生产环境部署后调度器停止使用开发服务器部署进程不稳定开发服务器 (app.run) 不适合生产环境进程崩溃或重启会导致调度器停止。生产环境应使用Gunicorn或uWSGI运行 Flask并使用Supervisor或systemd单独管理一个常驻的调度器进程。7.2 生产环境部署建议开发环境运行顺利不代表生产环境就高枕无忧。以下是为生产部署必须考虑的几点分离调度器与 Web 服务问题在app.py中启动调度器的方式使得调度器和 Web 服务耦合在同一个进程。当使用多 worker 模式如 Gunicorn部署时每个 worker 都会启动自己的调度器导致任务重复执行。方案将scheduler.py改造成一个独立的脚本如run_scheduler.py使用python run_scheduler.py单独运行。然后使用进程管理工具如 Supervisor来同时管理 Flask 应用进程和调度器进程。配置外置化将数据库连接字符串、密钥、告警 Webhook 地址等敏感信息从代码中移除改为从环境变量或配置文件如.env中读取。可以使用python-dotenv库。添加告警功能在monitor.py的probe_dns_target函数中当record.status不是success时触发告警逻辑如发送邮件、调用钉钉/企业微信机器人 Webhook。注意要添加告警抑制机制避免短时间内重复告警。数据库优化ProbeRecord表会随时间快速增长。需要定期归档或清理旧数据。可以在该表上按created_at字段建立分区或者定期执行清理任务。日志记录将print语句替换为标准的日志记录如 Pythonlogging模块并配置日志级别、格式和输出位置文件、syslog等便于问题追踪。7.3 利用 AI 编程助手的高效协作心得在开发过程中AI 助手能极大提升效率但也可能“翻车”。以下是一些实践建议明确需求在向 AI 提问前自己先理清功能边界、输入输出和异常情况。模糊的指令会得到模糊甚至错误的代码。分步验证不要一次性让 AI 生成全部代码。应该按模块如模型定义、探测函数、API 路由分别生成和测试。每完成一个可运行的小单元就验证一次。理解代码AI 生成的代码一定要读明白再使用。特别是涉及数据库操作、网络请求、并发处理的部分要清楚其原理和潜在风险。处理“幻觉”AI 可能会使用不存在的库函数或错误的参数。遇到报错时要优先查阅官方文档进行核实和修正。迭代优化第一版代码通常只解决“有无”问题。在跑通基础功能后可以再让 AI 协助进行代码重构、添加错误处理、编写单元测试等。通过这个项目我们不仅实现了一个可用的 DNS 监控面板更实践了一条从需求分析、技术选型、编码实现、问题排查到生产准备的完整开发路径。你可以在此基础上继续扩展告警、多用户、更丰富的图表分析等功能使其更贴合你的实际运维场景。