摘要：在金融、涉密政企及工业核心内网中，系统通常部署于与外网完全物理隔离（Air-Gapped Network）的私有云环境。在此环境下，传统的公有云短信、微信/飞书 Webhook 以及 SaaS 告警推送完全失效。当高可用集群（如 Keepalived、Pacemaker）发生脑裂（Split-Brain）或 VIP（虚拟 IP）漂移异常时，现场运维人员很难第一时间获知。本文将介绍如何基于 Python 编写轻量级 VIP 监听与脑裂检测服务，通过局域网 REST API 与 HMAC 安全签名驱动嵌入式声光终端，实现全彩 RGB 视觉矩阵与离线 TTS 语音播报，打造物理隔离环境下的高可靠现场告警防线。

一、 物理隔离网络（Air-Gapped）下的告警流转架构

在无法连接公有云 API 的私有内网中，告警系统必须做到完全本地化、低依赖与高自治：

+---------------------------------------------------------------+
|                私有云高可用集群 (Keepalived / VIP)             |
|  - Master / Backup 节点状态心跳监测                            |
|  - 状态切换事件: MASTER -> BACKUP / FAULT                     |
|  - 脑裂检测 (Split-Brain Detection)                           |
+-------------------------------+-------------------------------+
                                |
                                | (局域网本地 Socket / Shell 触发)
                                v
+---------------------------------------------------------------+
|               内网告警桥接服务 (Python Daemon)                 |
|  - 监听 VIP 漂移与状态机变化                                   |
|  - 本地 HMAC-SHA256 签名计算与时间戳防重放                      |
|  - 滑动窗口去重 (Debounce)                                     |
+-------------------------------+-------------------------------+
                                |
                                | (局域网内网 REST API / 纯 HTTP)
                                v
+---------------------------------------------------------------+
|                   嵌入式声光告警终端                           |
|  - RGB 全彩 LED 视觉矩阵 (常亮/闪烁/呼吸)                      |
|  - 本地离线 TTS 语音合成芯片 (无需外网 API，硬件解码)           |
+---------------------------------------------------------------+

核心设计原则：

1. 零外网依赖：告警终端内置硬件级离线 TTS 语音合成芯片，无需调用任何公有云 TTS 接口。

2. 状态双向感应：不仅在节点发生 FAULT（故障）时报警，在状态恢复（MASTER 复位）时同样通过绿灯常亮完成视觉与语义闭环。

二、 核心代码实现：Keepalived 告警 Notify 脚本与驱动服务

以下为部署在 Keepalived 节点上的状态监听与声光驱动 Python 脚本：

Python

import sys
import time
import json
import requests
import hashlib
import hmac

# 配置参数（纯局域网 IP）
ALARM_DEVICE_IP = "192.168.10.200"
API_KEY = "airgapped_cluster_monitor"
SECRET_KEY = "YourHMACSecretKey2026"

def calc_signature(timestamp, payload_str):
    message = f"{timestamp}\n{payload_str}".encode('utf-8')
    return hmac.new(SECRET_KEY.encode('utf-8'), message, hashlib.sha256).hexdigest()

def push_physical_alarm(tts_text, state_level):
    url = f"http://{ALARM_DEVICE_IP}/api/v1/send_msg"
    timestamp = str(int(time.time()))

    # 依据高可用节点状态映射视觉与听觉参数
    if state_level == "FAULT":
        color = "#FF0000"     # 故障：红色爆闪
        light_mode = "flash"
        audio_mode = "cycle"
        repeat_times = 3
    elif state_level == "BACKUP":
        color = "#FFA500"     # 切换备份：橙色呼吸
        light_mode = "breath"
        audio_mode = "once"
        repeat_times = 1
    elif state_level == "MASTER":
        color = "#00FF00"     # 恢复主节点：绿色常亮
        light_mode = "steady"
        audio_mode = "once"
        repeat_times = 1
    else:
        color = "#00FFFF"
        light_mode = "steady"
        audio_mode = "once"
        repeat_times = 1

    payload = {
        "text": tts_text,
        "color": color,
        "light_mode": light_mode,
        "audio_mode": audio_mode,
        "repeat_times": repeat_times
    }

    payload_str = json.dumps(payload, separators=(',', ':'))
    signature = calc_signature(timestamp, payload_str)

    headers = {
        "Content-Type": "application/json",
        "X-API-Key": API_KEY,
        "X-Timestamp": timestamp,
        "X-Signature": signature
    }

    try:
        resp = requests.post(url, data=payload_str, headers=headers, timeout=3)
        if resp.status_code == 200:
            print(f"[Success] 物理现场声光已响应: {tts_text}")
    except Exception as e:
        print(f"[Error] 局域网告警终端通信失败: {e}")

if __name__ == "__main__":
    # 接收来自 Keepalived notify 调用的参数
    # Keepalived 传递参数顺序: $1=GROUP|INSTANCE, $2=NAME, $3=STATE
    if len(sys.argv) < 4:
        print("Usage: python keepalived_notify.py GROUP|INSTANCE NAME STATE")
        sys.exit(1)

    type_name = sys.argv[1]
    instance_name = sys.argv[2]
    target_state = sys.argv[3].upper() # MASTER, BACKUP, FAULT

    if target_state == "FAULT":
        tts_message = f"私有云集群警报：实例 {instance_name} 进入故障状态，虚拟 IP 已漂移，请检查主节点"
    elif target_state == "BACKUP":
        tts_message = f"集群状态变更：实例 {instance_name} 已切换为备用节点"
    elif target_state == "MASTER":
        tts_message = f"集群恢复正常：实例 {instance_name} 已接管成为主节点"
    else:
        tts_message = f"集群状态未知变更：实例 {instance_name} 当前状态 {target_state}"

    print(f"[Keepalived Event] {tts_message}")
    push_physical_alarm(tts_message, target_state)

三、 Keepalived 配置文件对接 (keepalived.conf)

在私有云节点的 keepalived.conf 中配置 notify 脚本回调：

代码段

vrrp_instance VI_1 {
    state MASTER
    interface eth0
    virtual_router_id 51
    priority 100
    advert_int 1

    authentication {
        auth_type PASS
        auth_pass Sec12345
    }

    virtual_ipaddress {
        192.168.10.100/24
    }

    # 状态切换时触发 Python 局域网声光驱动脚本
    notify_master "/usr/bin/python3 /etc/keepalived/keepalived_notify.py INSTANCE VI_1 MASTER"
    notify_backup "/usr/bin/python3 /etc/keepalived/keepalived_notify.py INSTANCE VI_1 BACKUP"
    notify_fault  "/usr/bin/python3 /etc/keepalived/keepalived_notify.py INSTANCE VI_1 FAULT"
}

四、 隔离环境下的运维最佳实践

1. 内网 API 鉴权防伪造： 在隔离局域网中，仍需防范内部误操作或跨网段扫描。客户端发送请求时必须携带 HMAC-SHA256 签名与时间戳，告警终端拒绝无签名或时间偏差大于 ±300 秒的明文请求。

2. 私有云机房视觉色彩映射：

   • 红色闪烁 (#FF0000)：FAULT 状态或脑裂异常，需要运维人员进入现场处置。

   • 橙色呼吸 (#FFA500)：BACKUP 备用状态，提示降级运行。

   • 绿色常亮 (#00FF00)：MASTER 主节点正常运作。

3. 分时段音量管理： 结合脚本内部的时间判断逻辑，非值守时段将 audio_mode 设置为 none，仅保留 LED 爆闪指示，做到“物理强提醒、环境零骚扰”。

五、 总结

通过 Keepalived Notify -> 内网签名适配器 -> 嵌入式离线声光终端 的全自动化链路，私有云隔离环境下的集群故障与 VIP 漂移事件得以实时转化为物理现场精准的视觉与听觉指示。这种去外网依赖的纯局域网软硬协同架构，极大提升了隔离网络数据中心与专网节点的现场运维敏捷度与安全保障水平。