Your Name
d744d7df8e
P0-4: install.py _write_env() 值加引号转义，防止含#$\等字符的密码破坏.env格式 P0-5: install.py _build_redis_url() URL编码redis密码，防止@:等字符破坏URL解析 P1-6: auth_service.py _decode_access_token() 验证exp/sub声明存在，拒绝畸形JWT P1-7: websocket.py + webssh.py WebSocket JWT验证增加updated_at检查，密码修改后令牌失效 P1-8: auth.py 无refresh_token的logout返回更明确的提示信息 P1-9: install.py create_admin增加_is_installed()检查，防止.env存在后重复创建 P1-10: servers.py server_stats() 改用SQL聚合查询，避免加载全部服务器对象 P1-11: heartbeat_flush.py 移除get_redis()永不返回None的死代码 P1-13: sync_engine_v2.py completed/failed计数器加asyncio.Lock防止并发竞态 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""Nexus — Heartbeat Flush Background Task
|
|
Every 10 minutes: batch-flush Redis heartbeat data to MySQL servers table.
|
|
MySQL only stores historical snapshots; frontend reads live data from Redis.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import update
|
|
|
|
from server.infrastructure.database.session import AsyncSessionLocal
|
|
from server.infrastructure.redis.client import get_redis
|
|
from server.domain.models import Server
|
|
|
|
logger = logging.getLogger("nexus.heartbeat_flush")
|
|
|
|
FLUSH_INTERVAL = 600 # 10 minutes (seconds)
|
|
REDIS_KEY_PREFIX = "heartbeat:"
|
|
REDIS_KEY_EXPIRE = 600 # 10 minutes TTL for Redis heartbeat keys
|
|
|
|
|
|
async def heartbeat_flush_loop():
|
|
"""Background task: every 10min, flush Redis heartbeat data to MySQL.
|
|
|
|
Flow:
|
|
1. Scan all heartbeat:* keys in Redis
|
|
2. For each key, extract server_id + heartbeat data
|
|
3. Batch UPDATE MySQL servers table (is_online, system_info, last_heartbeat, agent_version)
|
|
4. Log count of flushed servers
|
|
"""
|
|
logger.info("Heartbeat flush loop started (interval: 10min)")
|
|
while True:
|
|
await asyncio.sleep(FLUSH_INTERVAL)
|
|
|
|
redis = get_redis()
|
|
|
|
# Scan all heartbeat:* keys
|
|
try:
|
|
keys = []
|
|
cursor = 0
|
|
while True:
|
|
cursor, batch = await redis.scan(cursor, match=f"{REDIS_KEY_PREFIX}*", count=100)
|
|
keys.extend(batch)
|
|
if cursor == 0:
|
|
break
|
|
except Exception as e:
|
|
logger.error(f"Redis scan failed: {e}")
|
|
continue
|
|
|
|
if not keys:
|
|
logger.info("No heartbeat keys in Redis — no agents reporting yet")
|
|
continue
|
|
|
|
flushed = 0
|
|
try:
|
|
async with AsyncSessionLocal() as session:
|
|
for key in keys:
|
|
try:
|
|
server_id = int(key.replace(REDIS_KEY_PREFIX, ""))
|
|
data = await redis.hgetall(key)
|
|
|
|
if not data:
|
|
continue
|
|
|
|
is_online = data.get("is_online") == "True"
|
|
system_info = data.get("system_info", "{}")
|
|
agent_version = data.get("agent_version", "")
|
|
|
|
# Parse last_heartbeat
|
|
last_hb_str = data.get("last_heartbeat", "")
|
|
try:
|
|
last_heartbeat = datetime.fromisoformat(last_hb_str)
|
|
except (ValueError, TypeError):
|
|
last_heartbeat = datetime.now(timezone.utc)
|
|
|
|
await session.execute(
|
|
update(Server)
|
|
.where(Server.id == server_id)
|
|
.values(
|
|
is_online=is_online,
|
|
system_info=system_info,
|
|
last_heartbeat=last_heartbeat,
|
|
agent_version=agent_version,
|
|
)
|
|
)
|
|
flushed += 1
|
|
except Exception as e:
|
|
logger.warning(f"Failed to flush server {key}: {e}")
|
|
# Don't rollback here — just skip this server and continue.
|
|
# A partial UPDATE on a single server won't corrupt others.
|
|
# The final commit() will persist all successful updates.
|
|
|
|
await session.commit()
|
|
logger.info(f"Heartbeat flush: {flushed}/{len(keys)} servers updated to MySQL")
|
|
except Exception as e:
|
|
logger.error(f"Heartbeat flush batch failed: {e}") |