Your Name
f9045a8404
- 将 Nexus/Nexus/* 移到仓库根目录（消除双层嵌套） - 删除旧的多层空壳目录 (server/, web/, agent/, deploy/, docs/) - 将PHP前端 (web/) 合并进Nexus仓库 — 统一管理 - 部署时 git clone Nexus 仓库即可，不再需要两个仓库 目录结构： server/ ← Python FastAPI后端 web/ ← PHP前端 (install.php, config.php, etc) deploy/ ← Supervisor + Shell健康检查 docs/ ← 部署文档 tests/ ← 测试 .env ← 不在git中 (install.php生成) .gitignore ← 排除 .env, SECRETS.md Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
98 lines
3.5 KiB
Python
98 lines
3.5 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
|
|
|
|
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 = await get_redis()
|
|
if not redis:
|
|
logger.warning("Redis unavailable — skip heartbeat flush")
|
|
continue
|
|
|
|
# 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.debug("No heartbeat keys in Redis — skip flush")
|
|
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.utcnow()
|
|
|
|
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.error(f"Failed to flush server {key}: {e}")
|
|
|
|
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}") |