实现一个内网服务监测告警系统

前言

昨天我的 pve 系统整个挂掉了，之前搭建的告警服务自然也死掉了，这就导致了我不能及时发现网站崩掉了，重启机器。

于是，我就把目光锁定到了家里的软路由上面，它是 x86 架构的，也安装了 docker ，我只需要用 python 写个脚本，做个 docker 服务即可。

功能设计

有了想法后，接下来需要先确定下要实现什么功能。

• 定时检查：每 N 秒检查一次指定主机的指定端口
• 自动告警：如果连续失败 N 次，就自动通过 QQ 邮箱发邮件通知
• Docker / docker-compose 支持：一个 docker-compose up -d 就搞定，不需要在宿主机安装什么复杂依赖
• 日志 + 时区：日志里记录访问时间 / 成功失败 / 告警状态，就算重启也能看到历史

实现过程

接下来就跟大家分享下我的具体实现过程。

• 用 Python + smtp + socket，做一个循环脚本：
  • 尝试 TCP connect （检测端口）
  • 连不上就计数，超过阈值就发邮件
• 用 Dockerfile 构建一个镜像，在里面安装 ping 、 ca-certificates ，配置时区，使得：
  • 容器里的时间符合预期
  • 脚本日志能实时输出，中断重启也方便查看
• docker-compose 管理：使用的时候只需要填写环境变量（目标主机 + 端口 + 邮箱 + 授权码…），然后 docker-compose up -d 就能全自动运行。

import os import smtplib import time import socket from email.mime.text import MIMEText from email.header import Header from email.utils import formataddr # 监控配置 TARGET_HOST = os.getenv("TARGET_HOST", "127.0.0.1") TARGET_PORT = int(os.getenv("TARGET_PORT", "80")) INTERVAL_SEC = int(os.getenv("INTERVAL_SEC", "60")) FAIL_THRESHOLD = int(os.getenv("FAIL_THRESHOLD", "3")) # 邮件配置（ QQ 邮箱） SMTP_HOST = os.getenv("SMTP_HOST", "smtp.qq.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER", "") SMTP_PASS = os.getenv("SMTP_PASS", "") MAIL_FROM = os.getenv("MAIL_FROM", SMTP_USER) MAIL_TO = os.getenv("MAIL_TO", "") def check_port(host: str, port: int, timeout=2) -> bool: """ 返回 True 表示端口可连接，False 表示失败 """ try: with socket.create_connection((host, port), timeout=timeout): return True except Exception: return False def send_mail(subject: str, content: str): if not (SMTP_HOST and SMTP_USER and SMTP_PASS and MAIL_TO): print("SMTP 配置不完整，无法发送邮件") return from_addr = MAIL_FROM or SMTP_USER msg = MIMEText(content, "plain", "utf-8") msg["From"] = formataddr(("Ping 告警系统", from_addr)) msg["To"] = formataddr(("告警接收人", MAIL_TO)) msg["Subject"] = Header(subject, "utf-8") print(f" [邮件] 准备连接 SMTP: host={SMTP_HOST}, port={SMTP_PORT}, user={SMTP_USER}") server = None try: if SMTP_PORT == 465: print(" [邮件] 使用 SMTP_SSL 连接（ 465 端口）") server = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10) else: print(" [邮件] 使用 SMTP + STARTTLS 连接") server = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10) server.ehlo() server.starttls() server.ehlo() server.login(SMTP_USER, SMTP_PASS) # sendmail 如果不抛异常，就认为成功 failed = server.sendmail(from_addr, [MAIL_TO], msg.as_string()) if failed: print(" [邮件] 部分收件人发送失败：", failed) else: print(" [邮件] 告警邮件已发送（ sendmail 返回正常）") except smtplib.SMTPResponseException as e: if e.smtp_code == -1 and e.smtp_error == b'\x00\x00\x00': print(" [邮件] QQ 在 QUIT 阶段返回 (-1, b'\\x00\\x00\\x00')，可忽略，邮件已经入队。") else: print(f" [邮件] SMTPResponseException：code={e.smtp_code}, error={e.smtp_error}") except Exception as e: print(f" [邮件] 发送失败：{repr(e)}，类型：{type(e)}") finally: if server is not None: try: server.quit() except Exception as e: # 这里的异常直接吞掉即可 print(f" [邮件] 关闭连接时异常（可忽略）：{repr(e)}") def main(): fail_count = 0 print( f"开始监控 {TARGET_HOST}:{TARGET_PORT}，每 {INTERVAL_SEC}s 检测一次，" f"连续失败 {FAIL_THRESHOLD} 次触发一次告警" ) while True: now = time.strftime("%F %T") ok = check_port(TARGET_HOST, TARGET_PORT) if ok: print(f"{now} [OK] {TARGET_HOST}:{TARGET_PORT} 端口可访问") if fail_count > 0: print(f"{now} 恢复正常，之前连续失败 {fail_count} 次，计数清零") fail_count = 0 else: fail_count += 1 print(f"{now} [FAIL] {TARGET_HOST}:{TARGET_PORT} 无法连接，连续失败次数：{fail_count}") if fail_count == FAIL_THRESHOLD: subject = f"[告警] {TARGET_HOST}:{TARGET_PORT} 无法访问" cOntent= ( f"目标 {TARGET_HOST}:{TARGET_PORT} 已连续 {FAIL_THRESHOLD} 次连接失败。\n" f"时间：{now}" ) send_mail(subject, content) time.sleep(INTERVAL_SEC) if __name__ == "__main__": main() 

构建与上传镜像

编写 DockerFile 镜像文件

FROM python:3.11-slim ENV TZ=Asia/Shanghai WORKDIR /app RUN apt-get update && \ apt-get install -y iputils-ping ca-certificates tzdata && \ ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && \ echo $TZ > /etc/timezone && \ update-ca-certificates && \ rm -rf /var/lib/apt/lists/* COPY ping_alert.py . CMD ["python", "-u", "ping_alert.py"] 

编写构建脚本

#!/usr/bin/env sh set -e # === 配置区：按需修改 === IMAGE_NAME="magiccoders/ping-alert" # magiccoders 需要改成你的 docker-hub 的用户名 TAG="latest" BUILD_COnTEXT="./app" # ======================= echo "==> 构建镜像: ${IMAGE_NAME}:${TAG}" docker build -t "${IMAGE_NAME}:${TAG}" "${BUILD_CONTEXT}" echo "==> 推送镜像到仓库: ${IMAGE_NAME}:${TAG}" docker push "${IMAGE_NAME}:${TAG}" echo "==> 完成：${IMAGE_NAME}:${TAG} 已发布" 

执行此脚本前，需要先在终端执行 docker login 命令登录到你的 docker-hub 账户。

编写 docker-compose 配置

构建好镜像后，需要创建docker-compose.yml文件来编排这个镜像运行所需的环境变量。

version: '3.8' services: ping-alert: image: magiccoders/ping-alert:latest # 此处就是存储在 docker-hub 上的镜像 container_name: ping-alert restart: always environment: # ===== 监控目标配置 ===== TARGET_HOST: "192.168.9.131" #监控目标机器 ip TARGET_PORT: "80" # 目标机器端口号 INTERVAL_SEC: "30" # 每 30 秒检查一次 FAIL_THRESHOLD: "3" # 连续 3 次失败发一封告警邮件 # ===== QQ 邮箱 SMTP 配置 ===== SMTP_HOST: "smtp.qq.com" SMTP_PORT: "465" SMTP_USER: "" # 你的 QQ 邮箱 SMTP_PASS: "" # 开通 SMTP 服务时得到的授权码 MAIL_FROM: "" # 和 SMTP_USER 保持一致 MAIL_TO: "" # 接受告警的邮箱 # 直接复用宿主机网络，方便访问内网 IP network_mode: "host" 

实现效果

我的软路由使用DPanel来管理 docker ，此处我就以它为例来讲解如何使用这个镜像。

如图所示，切换到 compose 选项卡，点击创建任务。

在打开的面板中，填写标识、名称，以及刚才的 docker-compose 配置代码，按需更改里面的变量即可

做完这些操作后，启动容器，查看日志，如果你的服务正常运行你就能看到如下所示的输出：

我把端口关闭，再来验证下失败的情况。

邮箱也收到了邮件。

最后，我启动服务，再来验证下他是否会清零计数。

项目地址

• GitHub 地址：ping-alert
• docker 镜像地址：magiccoders/ping-alert

写在最后

至此，文章就分享完毕了。

我是神奇的程序员，一位前端开发工程师。

如果你对我感兴趣，请移步我的个人网站，进一步了解。