a2s_monitor/test_dir/monitor_server.py

# coding:utf-8
import json
import os
import time
from mon import WorkerNodes
from loguru import logger
from multiprocessing import Process
import psutil

ServerMonitor = {}


def add_server(config_path):
    """
    添加服务器
    :return:
    """
    try:
        worker = WorkerNodes(config_path)
        worker.start()
        return None
    except Exception as e:
        logger.warning(f"{config_path}::启动失败", e)
        return False


def config_changed(config_dir):
    """
    检查配置文件的变动
    """
    for file in os.listdir(config_dir):
        file_path = os.path.join(config_dir, file)
        if os.path.isfile(file_path):
            if file_path.endswith("_nats.json"):
                config_value = json.load(open(file_path, "r"))
                name = config_value["name"]
                is_run = config_value.get("is_run", True)

                # 关闭服务
                if name in ServerMonitor:
                    if not is_run:
                        ServerMonitor[name][0].terminate()
                        del ServerMonitor[name]
                        logger.warning("关闭{}服务成功".format(name))
                    continue

                # 启动服务过滤关闭的服务
                if not is_run:
                    continue
                p = Process(target=add_server, args=(file_path,))
                p.start()
                ServerMonitor[name] = (p, p.pid)


def check_server():
    # 意外退出检查
    global ServerMonitor
    exists_pid = []
    for name, (_, p) in ServerMonitor.items():
        if not psutil.pid_exists(p):
            logger.warning(f"{name}::a2s监控服务意外退出，请尽快处理或重启")
            exists_pid.append(name)
    for name in exists_pid:
        del ServerMonitor[name]


def run_server(config_dir_path):
    while True:
        config_changed(config_dir_path)
        time.sleep(10)
        check_server()
        time.sleep(20)


if __name__ == '__main__':
    run_server("./docs")