Nexus/server/background/self_monitor.py

"""Nexus — Python Self-Monitor Background Task
Every 30 seconds: check Redis, MySQL, WebSocket connections.
If critical service fails → send Telegram system alert.
If service recovers → send Telegram recovery notification.
"""

import asyncio
import logging
import time

from sqlalchemy import text

from server.infrastructure.database.session import AsyncSessionLocal
from server.infrastructure.redis.client import get_redis
from server.infrastructure.telegram import send_telegram_system_alert
from server.api.websocket import manager as ws_manager, sync_manager as ws_sync_manager
from server.config import settings

logger = logging.getLogger("nexus.self_monitor")

MONITOR_INTERVAL = 30  # seconds
_TELEGRAM_COOLDOWN = 300  # 5 minutes — same alert won't re-push Telegram

# Track previous state for recovery detection
_prev_redis_ok = True
_prev_mysql_ok = True
# Track last Telegram push time per alert key (prevent spam)
_last_telegram_time: dict = {}


async def self_monitor_loop():
    """Background task: every 30s, self-check critical services.

    Checks:
    1. Redis connection — if lost, send Telegram warning; if recovered, send recovery
    2. MySQL connection — if lost, send Telegram warning; if recovered, send recovery
    3. WebSocket client count (via ConnectionManager)
    """
    global _prev_redis_ok, _prev_mysql_ok

    def _should_send(key: str) -> bool:
        """Rate-limit Telegram pushes to 1 per 5 minutes per alert key."""
        now = time.monotonic()
        last = _last_telegram_time.get(key, 0)
        if now - last < _TELEGRAM_COOLDOWN:
            return False
        _last_telegram_time[key] = now
        return True

    logger.info("Self-monitor loop started (interval: 30s)")
    while True:
        await asyncio.sleep(MONITOR_INTERVAL)

        # ── Redis check ──
        redis_ok = True
        try:
            redis = get_redis()
            await redis.ping()
        except Exception:
            redis_ok = False
            logger.warning("Redis health check failed", exc_info=True)
            try:
                if settings.REDIS_URL and _should_send("NOTIFY_SYSTEM_REDIS"):
                    await send_telegram_system_alert(
                        summary="Redis 连接失败，无法 ping 通",
                        title="Redis 连接异常",
                        notify_key="NOTIFY_SYSTEM_REDIS",
                        icon="🔴",
                        details=[
                            "检测来源：Layer 2 self_monitor（30 秒周期）",
                            "影响范围：子机实时心跳、在线状态、Redis 缓存读写可能中断",
                            "数据后果：前端仪表盘实时数据可能停滞，历史数据仍可从 MySQL 读取",
                        ],
                        action="请检查 1Panel Redis 容器状态与 NEXUS_REDIS_URL 配置",
                    )
            except Exception:
                logger.error("Failed to send Redis alert via Telegram")

        # Redis recovery notification
        if redis_ok and not _prev_redis_ok:
            try:
                await send_telegram_system_alert(
                    summary="Redis 连接已恢复，ping 检测通过",
                    title="Redis 连接恢复",
                    notify_key="NOTIFY_SYSTEM_REDIS",
                    icon="🟢",
                    details=[
                        "检测来源：Layer 2 self_monitor（30 秒周期）",
                        "当前状态：Redis 可正常读写，实时心跳应恢复更新",
                    ],
                )
                _last_telegram_time.pop("NOTIFY_SYSTEM_REDIS", None)  # clear cooldown
            except Exception:
                logger.error("Failed to send Redis recovery via Telegram")
            logger.info("Redis connection recovered")
        _prev_redis_ok = redis_ok

        # ── MySQL check ──
        mysql_ok = True
        try:
            async with AsyncSessionLocal() as session:
                await session.execute(text("SELECT 1"))
        except Exception:
            mysql_ok = False
            logger.warning("MySQL health check failed", exc_info=True)
            try:
                if _should_send("NOTIFY_SYSTEM_MYSQL"):
                    await send_telegram_system_alert(
                        summary="MySQL 连接失败，SELECT 1 探测未通过",
                        title="MySQL 数据库异常",
                        notify_key="NOTIFY_SYSTEM_MYSQL",
                        icon="🔴",
                        details=[
                            "检测来源：Layer 2 self_monitor（30 秒周期）",
                            "影响范围：设置、子机列表、审计、推送任务等所有 DB 读写",
                            "严重程度：平台核心功能可能不可用",
                        ],
                        action="请检查 1Panel MySQL 容器、磁盘空间与 DATABASE_URL 配置",
                    )
            except Exception:
                logger.error("Failed to send MySQL alert via Telegram", exc_info=True)

        # MySQL recovery notification
        if mysql_ok and not _prev_mysql_ok:
            try:
                await send_telegram_system_alert(
                    summary="MySQL 连接已恢复，SELECT 1 探测通过",
                    title="MySQL 数据库恢复",
                    notify_key="NOTIFY_SYSTEM_MYSQL",
                    icon="🟢",
                    details=[
                        "检测来源：Layer 2 self_monitor（30 秒周期）",
                        "当前状态：数据库读写应已恢复正常",
                    ],
                )
                _last_telegram_time.pop("NOTIFY_SYSTEM_MYSQL", None)  # clear cooldown
            except Exception:
                logger.error("Failed to send MySQL recovery via Telegram")
            logger.info("MySQL connection recovered")
        _prev_mysql_ok = mysql_ok

        # ── Log status ──
        logger.debug(
            f"Self-monitor: redis={'ok' if redis_ok else 'error'}, "
            f"mysql={'ok' if mysql_ok else 'error'}, "
            f"ws_alert_clients={ws_manager.client_count}, "
            f"ws_sync_clients={ws_sync_manager.client_count}"
        )