# coding:utf-8 from a2s.a2s_monitor import watch_monitor from utils.send_message import send_weixin import time from math import ceil, floor from utils.tools import calculate_file_hash import threading import json from utils.ecs import EcsClient from pynats import NATSClient from a2s.tools import json_serialize # 负载计算 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").replace(" ", "") 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.ecs_config["instance_name"] = self.worker_name self.change_alert = config.get("change_alert", True) self.load_alert = config.get("load_alert", True) self.load_alert_interval = config.get("load_alert_interval", 300) self.release_freeze_interval = config.get("release_freeze_interval", 60 * 30) self.weixin_bot_url = config.get("weixin_bot_url", "") self.refresh_instance_interval = config.get("refresh_instance_interval", 300) self.worker_timeout = config.get("worker_timeout", 300) self.monitor_subject = config.get("monitor_subject", "monitorRelease") self.nats_url = config.get("nats_url", "nats://172.17.4.188:19090") self.instances = [] self.release_instances = [] self._routine = [] self.last_update_time = 0 self.last_load_alert_time = 0 self.last_refresh_instance_time = 0 self.release_freeze_time = 0 self.ecs = EcsClient(self.ecs_config) 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 safe_logout(self, hosts): with NATSClient(self.nats_url) as nc: nc.connect() for host in hosts: nc.publish(subject=self.monitor_subject, payload=json_serialize({"host": host})) def refresh_instance(self): if not self.instances: return instances = [] # 分页查询状态 _instances = [] # 正常的机器 # 检查状态 for ids, ip in self.instances: instances.append(ids) if len(instances) > 5: instances_ids = self.ecs.instance_status(instances) if instances_ids: _instances.extend(instances_ids) instances = [] if instances: instances_ids = self.ecs.instance_status(instances) _instances.extend(instances_ids) # 无异常 if len(_instances) == len(self.instances): self.ecs_warning(f"实例状态更新成功,没有设备被释放！") return # 有异常 for instance in self.instances: if instance not in _instances: self.ecs.release_instance(instance[0], force=True) self.ecs_warning(f"实例状态更新成功,有{len(self.instances) - len(_instances)}台设备被释放！") # 更新存储 self.instances = _instances def loop_monitor(self): """ 循环检查 :return: """ while self._run: try: this_time = time.time() 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.pop(0) time.sleep(60) continue # 计算当前负载 load = current_nodes / (len(self.instances) * self.one_instance_node_num) self._routine.append(load) # 检查ecs的状态，去除系统自己释放的实例 if this_time - self.last_refresh_instance_time > self.refresh_instance_interval: self.refresh_instance() self.last_refresh_instance_time = this_time # 清理大于"最大观察时间"的负载 if len(self._routine) > max_number: self._routine.pop(0) # 冻结时间 if this_time - self.last_update_time < self.freeze_time: time.sleep(self.watch_interval) continue # 检查当前负载 if this_time - self.last_load_alert_time > self.load_alert_interval: self.ecs_warning( f"[{self.worker_name}] 当前负载({current_nodes}::{load})为节点数：：{self.get_node_num()}，实例数：：{self.get_instance_num()}") self.last_load_alert_time = this_time # 检查是否新增 if len(self._routine) >= add_watch_number: add_start_index = len(self._routine) - 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)) if self.change_alert: self.ecs_warning(f"[{self.worker_name}]新增，当前负载：{add_load}，预期新增：{num}台") self.add_instances(num) self.release_freeze_time = time.time() continue # 检查是否释放 if len(self._routine) >= release_watch_number and time.time() - self.release_freeze_time > self.release_freeze_interval: 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: # 释放机器数量 # 负载小于 0.1 if release_load < 0.1: num = 3 if len(self.instances) > 70 else 2 # 负载大于 0.6 else: num = 2 if len(self.instances) > 10 else 1 # 告警 if self.change_alert: self.ecs_warning(f"[{self.worker_name}]释放，当前负载：{release_load}，预期释放：{num}台") self.reduce_instances(num) self.release_freeze_time = time.time() except Exception as e: self.ecs_warning(f"警告：[{self.worker_name}]服务出现异常，尽快处理{e}") time.sleep(self.watch_interval) self.close_server_instances() def add_instances(self, num): """ 新增设备 :param num: :return: """ if num < 1: 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) # 开始新增，最大每次增加50台 if num > 50: num = 50 total_instance_ids = [] # 新增实例，分页5台 for i in range(floor(num / 5)): instance_ids = self.ecs.add_instance(5) if instance_ids: total_instance_ids.extend(instance_ids) self.instances.extend(instance_ids) self.last_update_time = time.time() else: self.ecs_warning(f"[{self.worker_name}]预期新增{num}设备，扩容失败！当前ECS:{len(self.instances)}台") # 不足5台 yu_num = num % 5 if yu_num > 0: instance_ids = self.ecs.add_instance(yu_num) if instance_ids: total_instance_ids.extend(instance_ids) self.instances.extend(instance_ids) self.last_update_time = time.time() else: self.ecs_warning(f"[{self.worker_name}]预期新增{num}设备，扩容失败！当前ECS:{len(self.instances)}台") # 通知 if self.change_alert: self.ecs_warning( f"[{self.worker_name}]新增{len(total_instance_ids)}(预期{num})台设备成功，当前ECS:{len(self.instances)}台") print(f"新增完成，结果：：{self.worker_name}::释放列表：{len(self.release_instances)}, 当前ECS列表:{self.instances}") else: # 通知 if self.change_alert: self.ecs_warning(f"[{self.worker_name}]ECS服务实例达到最大值:{self.max_instance}，预期新增{num}台, 扩容失败。") self.last_update_time = time.time() 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 # 开始释放操作 release_instance_ids = self.instances[:num] self.instances = self.instances[num:] hosts = [] for instance_id, ip in release_instance_ids: hosts.append(ip) self.release_instances.append([instance_id, ip, int(time.time())]) self.safe_logout(hosts) print(f"{self.worker_name}::释放列表：{self.release_instances}, 当前ECS列表:{self.instances}") self.ecs_warning(f"[{self.worker_name}]预计释放:{num}台实例，当前释放队列{len(self.release_instances)},预计剩余{len(self.instances)}") def loop_release_instances(self): """ 释放实例 :return: """ while self._run: time.sleep(60) delete_ids = [] for instance_detail in self.release_instances: instance_id, ip, t = instance_detail if int(time.time()) - t > self.worker_timeout: try: print(f"时间够了，开始释放{instance_id},时间IP{ip}:{t}") result = self.ecs.release_instance(instance_id, force=True) if result: delete_ids.append(instance_detail) except Exception as e: print(e) time.sleep(2) else: break try: for instance_detail in delete_ids: self.release_instances.remove(instance_detail) except Exception as e: self.ecs_warning(f"释放失败，请检查{e}") print(f"释放完成，结果：：{self.worker_name}::释放列表：{self.release_instances}, 当前ECS列表:{self.instances}") def ecs_warning(self, text): send_weixin(text, self.weixin_bot_url) def __initialize(self): if len(self.instances) < self.min_instance: while True: try: self.add_instances(self.min_instance - len(self.instances)) break except Exception as e: print(e) print("申请设备失败，等待60秒后重试") time.sleep(60) def start(self): t1 = threading.Thread(target=self.loop_monitor) t1.start() t2 = threading.Thread(target=self.update_config) t2.start() t3 = threading.Thread(target=self.loop_release_instances) t3.start() t1.join() t2.join() t3.join() self.close_server_instances() def update_config(self): """ 更新配置 :param config: :return: """ while self._run: config_md5 = calculate_file_hash(self.config_path) if config_md5 != self.config_md5: self.config_md5 = config_md5 config = json.load(open(self.config_path, "r")) self.worker_name = config.get("name", "worker").replace(" ", "") 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.freeze_time = config.get("freeze_time", 60) self._run = config.get("is_run", True) # 调整配置文件可以关闭服务 self.change_alert = config.get("change_alert", True) self.load_alert = config.get("load_alert", True) self.load_alert_interval = config.get("load_alert_interval", 300) self.weixin_bot_url = config.get("weixin_bot_url", "") self.release_freeze_interval = config.get("release_freeze_interval", 60 * 30) self.refresh_instance_interval = config.get("refresh_instance_interval", 300) self.worker_timeout = config.get("worker_timeout", 300) self.monitor_subject = config.get("monitor_subject", "monitorRelease") # 每间隔一段时间，查看一下配置文件的状态 time.sleep(10) def close_server_instances(self): release_ids = [] for ids, ip in self.instances: release_ids.append(ids) if self.release_instances: for instance_id, ip, t in self.release_instances: release_ids.append(instance_id) print("释放实例", release_ids) release_false = self.ecs.delete_instance(release_ids) # 释放完成 if release_false: if self.change_alert: self.ecs_warning(f"[{self.worker_name}]]服务关闭，部分实例释放失败，请手动释放,问题实例：{release_false}") else: self.ecs_warning(f"[{self.worker_name}]服务关闭，所有实例释放完成")