Your Name
96 lines
3.5 KiB
Python
96 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 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
|
|
skipped = 0
|
|
for key in keys:
|
|
try:
|
|
async with AsyncSessionLocal() as session:
|
|
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 — ensure timezone-aware
|
|
last_hb_str = data.get("last_heartbeat", "")
|
|
try:
|
|
last_heartbeat = datetime.fromisoformat(last_hb_str)
|
|
if last_heartbeat.tzinfo is None:
|
|
last_heartbeat = last_heartbeat.replace(tzinfo=timezone.utc)
|
|
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,
|
|
)
|
|
)
|
|
await session.commit()
|
|
flushed += 1
|
|
except Exception as e:
|
|
logger.warning(f"Failed to flush server {key}: {e}")
|
|
skipped += 1
|
|
|
|
logger.info(f"Heartbeat flush: {flushed}/{len(keys)} servers updated to MySQL" +
|
|
(f", {skipped} failed" if skipped else "")) |