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

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

• 请不要在回答技术问题时复制粘贴 AI 生成的内容

ChatGPT Image 2025 年 12 月 9 日 14_27_46

前言

昨天我的 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"