NAS电力监控程序


前几天已经完成了UPS的改造,现在已经把UPS使用起来了。虽然山特MT1000有控制软件,只是我就是懒,我不想去装这个软件,也懒得研究。现在NAS需要和UPS配合起来那就得自己写个程序来搞定了。

我先贴个简单得拓扑图,说一下我自己的环境,以及借助拓扑图说一下我的思路。

我的环境里有两个自己搭建的DNS服务器,两个DNS就是用研扬UP2的板子搭的,其中还有Chrony授时服务,这两个设备通过keepalived做了高可用。因为之前有些实验要做,涉及到网络隔离等,所以买了二手的交换机和防火墙。

下面说下具体思路:UPS为NAS、交换机、防火墙供电。电力监控程序定时监控两个地址,一个互联网地址,另一个是内网中不受UPS保护的设备。只有当这两个设备都检测到不通的时候才认为电力中断,开始UPS供电,程序控制NAS在指定的时间内进行关机。

NAS系统环境: 操作系统:debian 12.11 python3版本:3.11.2

附上程序源码:

#!/usr/bin/env python3
#-*- coding:utf-8 -*-
"""程序名称:electricity_monitor
功能:电力监控程序,用于监控市电网络,当市电断电UPS开始工作后及时关闭操作系统。
版本:v1.1.3
最后修改时间:2025年7月26日
最后修改内容:
1、优化第69行判断逻辑,因为ping失败后返回结果可能为不为1。
"""
import os
import sys
import time
import datetime
import subprocess

PERSONAL_PARAMETER_DIC = {  # 自定义参数字典。
    'op_time': '120',   # 关闭电源操作时间,单位:秒。
    'version': '1.1.3',   # 软件版本。
    'icmp_pack_count': '3',   # ICMP测试报文数量。
    'icmp_check_timeout': '1',  #ICMP测试超时时间,单位:秒。
    'check_addr_list': ['ns1.huaweicloud-dns.com', '192.168.1.1'],  # 定义全局变量,用于记录检查地址。第一个地址建议写互联网域名,第二个地址建议写内网不接入UPS设备的地址。
    'log_path': '/var/log/electricity_monitor.log',  # 日志路径。
    'last_log_print_time': 10  # 最后一次日志输出时间,单位:秒。
    }

NECESSARY_PARAMETER_DIC = {  # 关键参数字典,谨慎修改!
    'icmp_false_count': 0,  # ICMP失败计数。
    'countdown': int(PERSONAL_PARAMETER_DIC['op_time']),  # 系统关闭倒计时。
    'run_time': 0,  # 函数运行时间。
    'cmd_exe_failed_count': 0,  #命令执行失败次数,用于解决极端情况命令执行失败后陷入死循环问题。
    'first_log_print_count': 0,  # 首次日志打印次数。
    'last_log_print_count': 0,  # 首次日志打印次数。
    'run_status': True,  # 运行状态,用于决定程序退出与否的变量。
    'icmp_status': True,  # 定义全局变量icmp_status,用于保存icmp检查状态。默认值:True。
    'poweroff_cmd': ['shutdown', '-h', 'now']  # 关闭电源命令。
    }

def icmp_check():
    """函数名:icmp_check
    功能:实现icmp检查。
    """
    global PERSONAL_PARAMETER_DIC
    global NECESSARY_PARAMETER_DIC
    NECESSARY_PARAMETER_DIC['run_time'] = 0  # 归零函数运行时间,防止时间无限累加。
    time_start = time.perf_counter()  # 记录代码开始运行时间。
    check_result = subprocess.run(
        ['ping', 
         '-c', 
         PERSONAL_PARAMETER_DIC['icmp_pack_count'], 
         '-W', 
         PERSONAL_PARAMETER_DIC['icmp_check_timeout'], 
         PERSONAL_PARAMETER_DIC['check_addr_list'][0]], 
         stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )  # 向外网目标地址发送指定次数次ICMP报文,并应用指定超时时间。
    if check_result.returncode == 0:  # 外网地址检查通过。
        NECESSARY_PARAMETER_DIC['icmp_status'] = True  # 重置icmp检查状态,为防止内网设备故障而导致误关机操作。
    else:  # 外网地址检查失败。
        check_result = subprocess.run(
            ['ping', 
             '-c', 
             PERSONAL_PARAMETER_DIC['icmp_pack_count'], 
             '-W', 
             PERSONAL_PARAMETER_DIC['icmp_check_timeout'], 
             PERSONAL_PARAMETER_DIC['check_addr_list'][1]], 
             stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )  # 向内网目标地址发送指定次数次ICMP报文,并应用指定超时时间。
        if check_result.returncode != 0:  # 内网地址检查失败。
            NECESSARY_PARAMETER_DIC['icmp_status'] = False
            NECESSARY_PARAMETER_DIC['icmp_false_count'] += 1
        else:
            NECESSARY_PARAMETER_DIC['icmp_status'] = True  # 重置icmp检查状态,防止电力恢复后状态依旧为Flase而导致误关机操作。
    time_end = time.perf_counter()  # 记录代码结束运行时间。
    NECESSARY_PARAMETER_DIC['run_time'] = round(time_end - time_start)  # 记录函数运行时间,对结果四舍五入到整数。

def auto_operation():
    """函数名:auto_operation
    功能:实现自动操作。
    """
    global PERSONAL_PARAMETER_DIC
    global NECESSARY_PARAMETER_DIC
    if NECESSARY_PARAMETER_DIC['icmp_status'] == False:  # ICMP检查状态为False,则表示市电断电,已开始使用UPS。需要及时关闭电源。
        NECESSARY_PARAMETER_DIC['countdown'] = NECESSARY_PARAMETER_DIC['countdown'] - NECESSARY_PARAMETER_DIC['run_time']  # 倒计时需要减去ICMP检查函数的运行时间。
        if NECESSARY_PARAMETER_DIC['first_log_print_count'] == 0:  # 限制此日志只输出一次。
            make_logs(" | WAR | Power outage, UPS enabled!\n", 'c')
            NECESSARY_PARAMETER_DIC['first_log_print_count'] += 1
        if NECESSARY_PARAMETER_DIC['icmp_false_count'] > 0:  # ICMP失败计数大于0时表示检查发现掉电。
            if NECESSARY_PARAMETER_DIC['countdown'] <= PERSONAL_PARAMETER_DIC['last_log_print_time']:
                if NECESSARY_PARAMETER_DIC['last_log_print_count'] == 0:  #最后一次日志打印完成后不会重复打印。
                    make_logs(" | WAR | Power off the system after " + str(NECESSARY_PARAMETER_DIC['countdown']) +" seconds!\n", 'c')
                NECESSARY_PARAMETER_DIC['last_log_print_count'] += 1
                time.sleep(NECESSARY_PARAMETER_DIC['countdown'])
                execute_cmd = subprocess.run(NECESSARY_PARAMETER_DIC['poweroff_cmd'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                if execute_cmd.returncode != 0:  # 下电命令执行失败。
                    NECESSARY_PARAMETER_DIC['cmd_exe_failed_count'] += 1  # 命令执行失败次数加1。
                    if NECESSARY_PARAMETER_DIC['cmd_exe_failed_count'] == 5:  # 命令执行失败5次。
                        make_logs(" | ERR | Power off command execute failed five times!\n", 'c')
                        NECESSARY_PARAMETER_DIC['run_status'] = False
    else:  #icmp检查状态为True。
        if NECESSARY_PARAMETER_DIC['icmp_false_count'] > 1:  #icmp失败计数大于1时,则表示市电恢复。
            NECESSARY_PARAMETER_DIC['icmp_false_count'] = 0  # 重置icmp失败计数,防止市电恢复后icmp失败计数不归零。
            NECESSARY_PARAMETER_DIC['first_log_print_count'] = 0  # 重置首次日志输出次数,防止市电恢复后首次日志输出次数不归零。
            make_logs(" | INF | Power is restored, shutdown is canceled!\n", 'c')
        else:
            NECESSARY_PARAMETER_DIC['countdown'] = int(PERSONAL_PARAMETER_DIC['op_time'])  # 重置倒计时,防止市电正常时倒计时为负。
    NECESSARY_PARAMETER_DIC['countdown'] -= 1

def make_logs(content, type):
    """函数名:make_logs \n
    功能:实现日志生成。\n
    参数:\n
    content,表示需要写入的内容;\n
    type,表示日志类型,h表示需要写入的内容是日志头,c表示需要写入的是真实的日志内容。
    """
    global PERSONAL_PARAMETER_DIC
    log_content = None
    if type == 'c':
        now_time = datetime.datetime.now()
        log_content = str(now_time) + content
    else:
        log_content = content
    with open(PERSONAL_PARAMETER_DIC['log_path'], 'a') as log_file:
        log_file.write(log_content)

def main():  #程序主入口。
    global PERSONAL_PARAMETER_DIC
    global NECESSARY_PARAMETER_DIC
    while NECESSARY_PARAMETER_DIC['run_status']:
        if os.path.exists(PERSONAL_PARAMETER_DIC['log_path']) == False:  #首次启动创建日志文件头内容。
            log_head_info = "===== electricity_monitor " + PERSONAL_PARAMETER_DIC['version'] + " =====\n"
            make_logs(log_head_info, 'h')  #写入日志头信息。
        else:
            icmp_check()
            auto_operation()
            time.sleep(1)
    sys.exit()

if __name__ == "__main__":
    main()

根据自己的需求修改PERSONAL_PARAMETER_DIC字典里的值。为了使程序能在后台一直监控,需要再写一个service文件。我没有使用crontab,我使用systemctl管理程序的启停。

附上service文件内容:

[Unit]
Description=electricity monitor
After=network.target

[Service]
Type=simple
ExecStart=python3 /opt/electricity_monitor/electricity_monitor.py
ExecStop=/usr/bin/kill electricity_monitor
User=root
RemainAfterExit=yes

[Install]
WantedBy=multi-user.target

将service文件保存到/usr/lib/systemd/system,命名为:electricity_monitor.service。往/etc/systemd/system创建一个软链。

ln -s /usr/lib/systemd/system/electricity_monitor.service /etc/systemd/system/emd.service

相关启停命令如下:

启动并加入开机自启:systemctl enable electricity_monitor --now
停止服务:systemctl stop emd
启动服务:systemctl start emd

现在这个程序已经在我自己的NAS跑起来了。以后有什么更新的话会另起文章并在这里加链接。