a2s_monitor/test_dir/mon.py

# coding:utf-8
from a2s.a2s_monitor import watch_monitor
import time
from math import ceil, floor
from file_server import add_file_process
from picture_server import add_picture_process
from utils.tools import calculate_file_hash
import threading
import json
import multiprocessing

func_dict = {
    "test2txt_add": add_file_process,
    "test2img_add": add_picture_process
}


# 负载计算
class WorkerNodes(object):
    def __init__(self, config_path):
        self.config_path = config_path
        self.config_md5 = calculate_file_hash(self.config_path)
        config = json.load(open(self.config_path, "r"))
        self.worker_name = config.get("name", "worker")
        self.one_instance_node_num = config.get("one_instance_node_num", 1)
        self.max_instance = config.get("max_instance", 5)
        self.min_instance = config.get("min_instance", 1)
        self.watch_interval = config.get("watch_interval", 5)
        self.add_watch_time = config.get("add_watch_time", 180)
        self.release_watch_time = config.get("release_watch_time", 1200)
        self.a2s_ip = config.get("a2s_ip", "")
        self.upper_threshold = config.get("upper_threshold", 0.8)
        self.down_threshold = config.get("down_threshold", 0.5)
        self.ecs_config = config.get("ecs_config", {})
        self.freeze_time = config.get("freeze_time", 60)
        self._run = config.get("is_run", True)  # 调整配置文件可以关闭服务
        self.instances = []
        self._routine = []
        self.last_update_time = 0
        self.__initialize()

    def get_instance_num(self):
        """
        获取实例数量
        :return:
        """
        return len(self.instances)

    def get_node_num(self):
        """
        获取节点数量
        :return:
        """
        return len(self.instances) * self.one_instance_node_num

    def loop_monitor(self):
        """
        循环检查
        :return:
        """
        while self._run:
            try:
                add_watch_number = ceil(self.add_watch_time / self.watch_interval)
                release_watch_number = ceil(self.release_watch_time / self.watch_interval)
                max_number = max([add_watch_number, release_watch_number])
                current_nodes = watch_monitor(self.a2s_ip, self.worker_name)
                if not isinstance(current_nodes, int):
                    self.ecs_warning("警告：A2S服务监控出现异常，获取当前请求数量失败，当前服务停止增减设备。尽快解决当前问题，避免服务中断！！！")
                    if self._routine:
                        self._routine.remove(0)
                    time.sleep(60)
                    continue

                # 获取当前负载
                load = current_nodes / len(self.instances) * self.one_instance_node_num
                self._routine.append(load)

                # 清理大于"最大观察时间"的负载
                if len(self._routine) > max_number:
                    self._routine.pop(0)
                if time.time() - self.last_update_time < self.freeze_time:
                    print(f"{self.worker_name}::冻结时间未到，不进行增减设备")
                    time.sleep(self.watch_interval)
                    continue
                # 检查是否新增
                if len(self._routine) >= add_watch_number:
                    add_start_index = len(self._routine) - add_watch_number
                    print("长度比：：：", len(self._routine), len(self._routine[add_start_index:]), add_watch_number)
                    add_load = sum(self._routine[add_start_index:]) / add_watch_number
                    if add_load > self.upper_threshold:
                        num = ceil((add_load - self.upper_threshold) * len(self.instances))
                        print(f"{self.worker_name}新增add_load：：：", add_load, self.upper_threshold, num)
                        self.add_instances(num)
                        self.last_update_time = time.time()
                        continue

                # 检查是否释放
                if len(self._routine) >= release_watch_number:
                    release_start_index = len(self._routine) - release_watch_number
                    release_load = sum(self._routine[release_start_index:]) / release_watch_number
                    if release_load < self.down_threshold:
                        num = floor((1 - self.down_threshold) * len(self.instances))
                        print(f"{self.worker_name}释放release_load：：：", release_load, self.down_threshold, num)
                        self.reduce_instances(num)
                        self.last_update_time = time.time()

                # 检查间隔
                print(
                    f"{self.worker_name} 当前负载({current_nodes}::{load})为节点数：：{self.get_node_num()}，实例数：：{self.get_instance_num()}")
            except Exception as e:
                print("baocuole==>", e)
            time.sleep(self.watch_interval)
        self.close_server_instances()

    def add_instances(self, num):
        """
        新增设备
        :param num:
        :return:
        """
        if len(self.instances) < self.max_instance:
            if self.max_instance < len(self.instances) + num:
                num = self.max_instance - len(self.instances)
            if len(self.instances) + num < self.min_instance:
                num = self.min_instance - len(self.instances)
            print(f"ECS新增{num}设备，当前ECS数量为{len(self.instances)}")
            for i in range(num):
                p = multiprocessing.Process(target=func_dict[f"{self.worker_name}_add"])
                p.start()
                self.instances.append(p)
            # 开始新增
        else:
            self.ecs_warning(f"ECS 扩容失败,{self.worker_name}服务已达到最大实例数量:{self.max_instance}")

    def reduce_instances(self, num):
        """
        减少设备
        :param num:
        :return:
        """
        if len(self.instances) > self.min_instance:
            if len(self.instances) < self.min_instance + num:
                num = len(self.instances) - self.min_instance
            # 开始释放操作
            for ind in range(num):
                if ind < num:
                    p = self.instances.pop(0)
                    print("ECS释放进程：", p, p.pid)
                    p.terminate()
        else:
            self.ecs_warning(f"ECS 缩减失败,{self.worker_name}服务已达到最小实例数量:{self.min_instance}")

    def ecs_warning(self, text):
        print("警告：：：⚠️", text)

    def __initialize(self):
        if len(self.instances) < self.min_instance:
            self.add_instances(self.min_instance - len(self.instances))

    def start(self):
        t1 = threading.Thread(target=self.loop_monitor)
        t1.start()
        t2 = threading.Thread(target=self.update_config)
        t2.start()
        t1.join()
        t2.join()

    def update_config(self):
        """
        更新配置
        :param config:
        :return:
        """
        while self._run:
            config_md5 = calculate_file_hash(self.config_path)
            if config_md5 != self.config_md5:
                config = json.load(open(self.config_path, "r"))
                self.worker_name = config.get("name", "worker")
                self.one_instance_node_num = config.get("one_instance_node_num", 1)
                self.max_instance = config.get("max_instance", 5)
                self.min_instance = config.get("min_instance", 1)
                self.watch_interval = config.get("watch_interval", 5)
                self.add_watch_time = config.get("add_watch_time", 180)
                self.release_watch_time = config.get("release_watch_time", 1200)
                self.a2s_ip = config.get("a2s_ip", "")
                self.upper_threshold = config.get("upper_threshold", 0.8)
                self.down_threshold = config.get("down_threshold", 0.5)
                self.ecs_config = config.get("ecs_config", {})
                self._run = config.get("is_run", True)  # 调整配置文件可以关闭服务
            # 每间隔一段时间，查看一下配置文件的状态
            time.sleep(10)

    def close_server_instances(self):
        for p in self.instances:
            p.terminate()