"""后台高负载告警任务：CPU/内存超阈值告警 + 定时状态推送。""" import asyncio import logging import time import monitor import templates log = logging.getLogger("alerter") def _top1_str(p: dict) -> str: return f"{p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB" class Alerter: def __init__(self, ob, cfg: dict): self.ob = ob self.groups: list = [] self.friends: list = [] self.apply_config(cfg, initial=True) self._cpu_hits = 0 self._mem_hits = 0 self._cpu_alerting = False self._mem_alerting = False self._cpu_last_ts = 0.0 self._mem_last_ts = 0.0 self._task = None def apply_config(self, cfg: dict, initial: bool = False): """热重载告警参数（保留告警状态）。""" if initial: self.interval = cfg.get("interval", 10) self.threshold = cfg.get("cpu_threshold", 85.0) self.consecutive_need = cfg.get("consecutive", 2) self.cooldown = cfg.get("cooldown", 600) self.top_n = cfg.get("top_n", 5) else: self.interval = cfg.get("interval", self.interval) new_threshold = cfg.get("cpu_threshold", self.threshold) if new_threshold != self.threshold: log.info("CPU告警阈值变更: %.0f%% -> %.0f%%", self.threshold, new_threshold) self.threshold = new_threshold self.consecutive_need = cfg.get("consecutive", self.consecutive_need) self.cooldown = cfg.get("cooldown", self.cooldown) self.top_n = cfg.get("top_n", self.top_n) self.memory_threshold = cfg.get("memory_threshold", self.__dict__.get("memory_threshold", 90.0)) self.groups = cfg.get("groups", self.groups) or [] self.friends = cfg.get("friends", self.friends) or [] def start(self): self._task = asyncio.get_running_loop().create_task(self._loop()) async def stop(self): if self._task: self._task.cancel() async def _loop(self): log.info("告警监控已启动: CPU阈值 %.0f%%, 内存阈值 %.0f%%, 连续 %d 次, 间隔 %ds, 冷却 %ds", self.threshold, self.memory_threshold, self.consecutive_need, self.interval, self.cooldown) while True: try: await asyncio.sleep(self.interval) if not self.ob.connected or (not self.groups and not self.friends): continue cpu = monitor.get_cpu() mem = monitor.get_memory() await self._check(cpu["percent"], mem["percent"]) except asyncio.CancelledError: break except Exception: log.exception("告警循环出错") async def _check(self, cpu_percent: float, mem_percent: float): now = time.monotonic() # CPU 告警 if cpu_percent >= self.threshold: self._cpu_hits += 1 if (self._cpu_hits >= self.consecutive_need and (not self._cpu_alerting or now - self._cpu_last_ts >= self.cooldown)): self._cpu_alerting = True self._cpu_last_ts = now await self._send_cpu_alert(cpu_percent) else: self._cpu_hits = 0 if self._cpu_alerting: self._cpu_alerting = False msg = templates.render("recovered", cpu=f"{cpu_percent:.1f}") log.info(msg) await self.ob.broadcast(msg, self.groups, self.friends) # 内存告警 if mem_percent >= self.memory_threshold: self._mem_hits += 1 if (self._mem_hits >= self.consecutive_need and (not self._mem_alerting or now - self._mem_last_ts >= self.cooldown)): self._mem_alerting = True self._mem_last_ts = now await self._send_mem_alert(mem_percent) else: self._mem_hits = 0 if self._mem_alerting: self._mem_alerting = False msg = templates.render("recovered_mem", mem=f"{mem_percent:.1f}") log.info(msg) await self.ob.broadcast(msg, self.groups, self.friends) async def _send_cpu_alert(self, cpu_percent: float): top = monitor.get_top_processes(self.top_n) cpu = monitor.get_cpu() load = cpu["load"] top_list = "\n".join( f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB" for i, p in enumerate(top, 1) ) top1 = top[0] if top else {"name": "无数据", "pid": "-", "cpu": 0, "mem_mb": 0} msg = templates.render( "alert", threshold=f"{self.threshold:.0f}", cpu=f"{cpu_percent:.1f}", cores=cpu["cores"], load=f"{load[0]:.2f} / {load[1]:.2f} / {load[2]:.2f}", top_n=self.top_n, top_list=top_list or "无数据", top1=_top1_str(top1), ) log.warning("触发CPU高负载告警: %.1f%%", cpu_percent) await self.ob.broadcast(msg, self.groups, self.friends) async def _send_mem_alert(self, mem_percent: float): m = monitor.get_memory() top1 = monitor.get_top_memory(1) top1 = top1[0] if top1 else {"name": "无数据", "pid": "-", "cpu": 0, "mem_mb": 0} msg = templates.render( "alert_mem", threshold=f"{self.memory_threshold:.0f}", mem=f"{mem_percent:.1f}", mem_detail=f"{m['used_gb']}/{m['total_gb']}GB (Swap {m['swap_percent']}%)", top1=_top1_str(top1), ) log.warning("触发内存高占用告警: %.1f%%", mem_percent) await self.ob.broadcast(msg, self.groups, self.friends) class Reporter: """定时向配置目标推送状态报告。""" def __init__(self, ob, cfg: dict): self.ob = ob self.enabled = cfg.get("enabled", False) self.interval_minutes = cfg.get("interval_minutes", 60) self.groups = cfg.get("groups", []) or [] self.friends = cfg.get("friends", []) or [] self._task = None def start(self): if not self.enabled: log.info("定时报告未启用") return self._task = asyncio.get_running_loop().create_task(self._loop()) async def stop(self): if self._task: self._task.cancel() async def _loop(self): log.info("定时报告已启动: 每 %d 分钟推送一次", self.interval_minutes) while True: await asyncio.sleep(self.interval_minutes * 60) try: if not self.ob.connected or (not self.groups and not self.friends): continue status = bot_status() msg = templates.render("report", status=status) await self.ob.broadcast(msg, self.groups, self.friends) log.info("已推送定时报告") except asyncio.CancelledError: break except Exception: log.exception("定时报告推送出错") def bot_status() -> str: """状态文本（避免 bot <-> alerter 循环导入，延迟导入 bot）。""" import bot return bot.fmt_status(monitor.get_status())