yumoqing
fa257f36b1
依据现有实现补全「自动识别故障 → 人工确认切换 → 切回」运维核心闭环。 触发模式 1A+2B+3A：VIP 浮动 + 半自动 + 切前隔离原主（防脑裂）。 【表结构】 clusters 加 vip/failover_mode；node_commands 加 command_type 新增 switchover_steps（切换步骤级审计） 【Agent 六动作（脚本化、幂等、结构化返回）】 fence(停MySQL+摘VIP)/unfence(恢复)/promote_master(STOP+RESET SLAVE) revert_promote(降回从库)/flip_sync(反转同步)/verify(可写+VIP校验) 【切换编排器 foms/failover.py】 六步状态机：precheck→fence→promote→flip_sync→verify→converge 失败 → 已完成步骤逆向回滚 → switchover_logs=failed 回滚也失败 → rollback_failed + 最高级告警冒泡人工 防脑裂不变量：fence 先于 promote，fence 失败硬中止绝不 promote 【健康判定器】 后台循环（默认5s）扫 active 集群，心跳超时/复制异常判定 防抖：连续 failover_threshold 次命中才判故障（防瞬时抖动误判） 半自动：只告警（写 switchover_logs + hostwatch_events）不切，等人确认 【切回 + 回滚语义】 trigger_switchback 反向六步；回滚用 unfence/revert_promote/反转/换回 【部署文档】 docs/deploy-operations.md：架构/三机部署/keepalived/Agent 环境变量/ 日常运维/失败语义/防脑裂/已知限制 验证：语法全过；编排器↔Agent 六命令类型对齐；六步顺序正确； 防脑裂不变量成立（fence idx=1 < promote idx=2）；注册链完整。
505 lines
23 KiB
Python
505 lines
23 KiB
Python
"""FOMS 切换编排器 — 六步状态机 + 逆向回滚 + 防脑裂不变量。
|
||
|
||
设计（2026-08-26 定调，1A+2B+3A）：
|
||
· 流量切换：VIP 浮动（keepalived 兜底）
|
||
· 触发模式：半自动（健康判定器告警 + 人工确认后切）
|
||
· 防脑裂：fence（隔离原主）先于 promote（立新主）；fence 失败硬中止
|
||
|
||
切换六步（failover A→B）：
|
||
step1 precheck 编排器本地：B 在线 + 心跳新鲜 + 复制延迟 < 阈值（防丢数据）
|
||
step2 fence 下发 A：停 MySQL + 摘 VIP ← 防脑裂关键，失败必中止
|
||
step3 promote 下发 B：STOP SLAVE + RESET SLAVE ALL（脚本化原子执行）
|
||
step4 flip_sync DB 层反转 filesync 方向 + 下发 B 确认
|
||
step5 verify 下发 B：连库读写 + VIP 归属检查（业务层留人工确认）
|
||
step6 converge 收敛：集群主备互换 + 节点角色互换 + 日志终态
|
||
|
||
任何一步 failed → 已完成步骤逆向回滚 → switchover_logs=failed；
|
||
回滚也失败 → rollback_failed + 最高级告警冒泡人工。
|
||
|
||
松耦合约束：
|
||
· precheck/flip_sync 读 dbbackup_replications / filesync_directories（同库只读）
|
||
· 告警写 hostwatch_events（与 hostwatch 模块约定的事件结构）
|
||
"""
|
||
import asyncio
|
||
import json
|
||
from datetime import datetime
|
||
from appPublic.uniqueID import getID
|
||
from appPublic.timeUtils import curDateString
|
||
from sqlor.dbpools import get_sor_context
|
||
from ahserver.serverenv import ServerEnv
|
||
|
||
DBNAME = 'foms'
|
||
|
||
# 幂等标志：防热重载重复启动健康判定器后台任务
|
||
_failover_registered = False
|
||
|
||
# ═══ 常量（防丢数据 / 超时 / 轮询）═══
|
||
MAX_LAG_SECONDS = 5 # precheck：复制延迟必须 ≤ 该值才允许切换（R4）
|
||
HEARTBEAT_FRESH_SEC = 60 # precheck：目标节点心跳须在 60s 内
|
||
CMD_TIMEOUT = 90 # 等待 Agent 命令完成的最长秒数
|
||
CMD_POLL_INTERVAL = 2 # 轮询命令结果间隔（切换期加密，缓解 R1 延迟）
|
||
JUDGE_INTERVAL = 5 # 健康判定器扫描间隔（秒）
|
||
|
||
|
||
def _now():
|
||
return curDateString()
|
||
|
||
|
||
def _parse_dt(v):
|
||
"""把库里取出的时间值解析成 datetime（兼容 str/datetime）。"""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, datetime):
|
||
return v
|
||
for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S'):
|
||
try:
|
||
return datetime.strptime(str(v)[:19], fmt)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
return None
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 步骤记录（switchover_steps 审计）
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def _record_step(sor, switchover_id, seq, step_name, status, output='', error=''):
|
||
"""写一条步骤审计记录（先查后建，保证幂等）。不写 None 值（sqlor 不接受）。"""
|
||
existing = await sor.sqlExe(
|
||
'SELECT id FROM switchover_steps WHERE switchover_id=${sw}$ AND seq=${seq}$',
|
||
{'sw': switchover_id, 'seq': seq})
|
||
if existing:
|
||
upd = {
|
||
'id': getattr(existing[0], 'id', ''),
|
||
'status': status,
|
||
'output': (output or '')[:2000],
|
||
'error': (error or '')[:2000],
|
||
}
|
||
if status in ('success', 'failed', 'skipped'):
|
||
upd['completed_at'] = _now()
|
||
await sor.U('switchover_steps', upd)
|
||
else:
|
||
rec = {
|
||
'id': getID(),
|
||
'switchover_id': switchover_id,
|
||
'seq': seq,
|
||
'step_name': step_name,
|
||
'status': status,
|
||
'output': (output or '')[:2000],
|
||
'error': (error or '')[:2000],
|
||
}
|
||
if status == 'running':
|
||
rec['started_at'] = _now()
|
||
await sor.C('switchover_steps', rec)
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# Agent 命令下发 + 等待（结构化类型）
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def _send_command(sor, node_id, command_type, command):
|
||
"""向节点下发结构化命令，返回 cmd_id。"""
|
||
cmd_id = getID()
|
||
await sor.C('node_commands', {
|
||
'id': cmd_id,
|
||
'node_id': node_id,
|
||
'command_type': command_type,
|
||
'command': command or command_type,
|
||
'status': 'pending',
|
||
'created_at': _now(),
|
||
})
|
||
return cmd_id
|
||
|
||
|
||
async def _wait_command(sor, cmd_id, timeout=CMD_TIMEOUT):
|
||
"""轮询 node_commands 直到终态，返回 (ok, exit_code, output)。"""
|
||
deadline = asyncio.get_event_loop().time() + timeout
|
||
while asyncio.get_event_loop().time() < deadline:
|
||
rows = await sor.sqlExe(
|
||
'SELECT status, exit_code, output FROM node_commands WHERE id=${cid}$',
|
||
{'cid': cmd_id})
|
||
if rows:
|
||
r = rows[0]
|
||
status = getattr(r, 'status', '')
|
||
if status in ('success', 'failed'):
|
||
return (status == 'success'), getattr(r, 'exit_code', -1), getattr(r, 'output', '')
|
||
await asyncio.sleep(CMD_POLL_INTERVAL)
|
||
return False, -1, f'command timeout after {timeout}s'
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 各步骤实现
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def _step_precheck(sor, ctx):
|
||
"""step1：本地预检 —— 目标节点在线、心跳新鲜、复制延迟达标（防丢数据）。"""
|
||
master_id, standby_id = ctx['from_node_id'], ctx['to_node_id']
|
||
target = await sor.R('nodes', {'id': standby_id})
|
||
if not target:
|
||
return False, '目标节点不存在'
|
||
t = target[0]
|
||
if getattr(t, 'status', '') != 'online':
|
||
return False, f"目标节点不在线（status={getattr(t, 'status', '')}）"
|
||
hb = _parse_dt(getattr(t, 'last_heartbeat', None))
|
||
if hb is None or (datetime.now() - hb).total_seconds() > HEARTBEAT_FRESH_SEC:
|
||
return False, '目标节点心跳不新鲜（超过 %ds）' % HEARTBEAT_FRESH_SEC
|
||
# 复制延迟检查（松耦合读 dbbackup 表）
|
||
repls = await sor.sqlExe(
|
||
"SELECT db_name, seconds_behind, sync_status FROM dbbackup_replications "
|
||
"WHERE cluster_id=${cid}$ AND enabled='1'", {'cid': ctx['cluster_id']})
|
||
for r in (repls or []):
|
||
lag = getattr(r, 'seconds_behind', None)
|
||
if lag is None:
|
||
lag = 999
|
||
if int(lag) > MAX_LAG_SECONDS:
|
||
return False, f"复制延迟过大：{getattr(r, 'db_name', '')} behind {lag}s（阈值 {MAX_LAG_SECONDS}s，防丢数据）"
|
||
return True, 'precheck passed'
|
||
|
||
|
||
async def _step_fence(sor, ctx):
|
||
"""step2：隔离原主（停 MySQL + 摘 VIP）。防脑裂关键，失败必中止。"""
|
||
cmd_id = await _send_command(sor, ctx['from_node_id'], 'fence', 'fence')
|
||
ok, code, out = await _wait_command(sor, cmd_id)
|
||
return ok, out
|
||
|
||
|
||
async def _step_promote(sor, ctx):
|
||
"""step3：提升新主（STOP SLAVE + RESET SLAVE ALL，脚本化原子）。"""
|
||
cmd_id = await _send_command(sor, ctx['to_node_id'], 'promote_master', 'promote_master')
|
||
ok, code, out = await _wait_command(sor, cmd_id)
|
||
return ok, out
|
||
|
||
|
||
async def _step_flip_sync(sor, ctx):
|
||
"""step4：DB 层反转 filesync 方向 + 下发目标节点确认。"""
|
||
from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
|
||
# 取原主/新主的 host，用于方向反转
|
||
nodes = {}
|
||
for nid in (from_id, to_id):
|
||
recs = await sor.R('nodes', {'id': nid})
|
||
if recs:
|
||
nodes[nid] = getattr(recs[0], 'host', '')
|
||
from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '')
|
||
# 反转：source_host 从原主指向新主（A→B 变 B→A）
|
||
await sor.sqlExe(
|
||
'UPDATE filesync_directories SET source_host=${th}$, dest_host=${fh}$, '
|
||
'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${fh}$',
|
||
{'th': to_host, 'fh': from_host, 'now': _now(), 'cid': ctx['cluster_id']})
|
||
cmd_id = await _send_command(sor, to_id, 'flip_sync', 'flip_sync')
|
||
ok, code, out = await _wait_command(sor, cmd_id)
|
||
return ok, f'flip db done; agent: {out}'
|
||
|
||
|
||
async def _step_verify(sor, ctx):
|
||
"""step5：校验新主可写 + VIP 归属（业务层留人工确认）。"""
|
||
cmd_id = await _send_command(sor, ctx['to_node_id'], 'verify', 'verify')
|
||
ok, code, out = await _wait_command(sor, cmd_id)
|
||
if not ok:
|
||
return False, out
|
||
return True, out + '（业务层请人工确认）'
|
||
|
||
|
||
async def _step_converge(sor, ctx):
|
||
"""step6：收敛 —— 集群主备互换 + 节点角色互换 + 集群恢复 running。"""
|
||
from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
|
||
await sor.U('clusters', {
|
||
'id': ctx['cluster_id'],
|
||
'master_node_id': to_id,
|
||
'standby_node_id': from_id,
|
||
'status': 'running',
|
||
'updated_at': _now(),
|
||
})
|
||
await sor.U('nodes', {'id': to_id, 'role': 'master', 'updated_at': _now()})
|
||
await sor.U('nodes', {'id': from_id, 'role': 'standby', 'updated_at': _now()})
|
||
return True, 'converged: master/standby swapped'
|
||
|
||
|
||
# 步骤表（正序）
|
||
STEPS = [
|
||
('precheck', _step_precheck),
|
||
('fence', _step_fence),
|
||
('promote', _step_promote),
|
||
('flip_sync', _step_flip_sync),
|
||
('verify', _step_verify),
|
||
('converge', _step_converge),
|
||
]
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 逆向回滚
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def _rollback(sor, ctx, completed_steps):
|
||
"""逆向回滚已完成步骤。返回 (ok, detail)。
|
||
|
||
回滚语义：
|
||
fence → 恢复原主（start MySQL + 还 VIP）
|
||
promote → 把新主降回从库（CHANGE MASTER 指向原主）——需原主已恢复
|
||
flip_sync → 反转回来
|
||
verify → 无需回滚
|
||
converge → 主备换回
|
||
"""
|
||
detail = []
|
||
ok_all = True
|
||
for step_name in reversed(completed_steps):
|
||
if step_name == 'fence':
|
||
# 恢复原主：start MySQL + 还 VIP（用 unfence 类型，Agent 幂等执行）
|
||
cmd_id = await _send_command(sor, ctx['from_node_id'], 'unfence', 'unfence')
|
||
ok, _, out = await _wait_command(sor, cmd_id)
|
||
detail.append(f'unfence {ctx["from_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}')
|
||
ok_all = ok_all and ok
|
||
elif step_name == 'promote':
|
||
# 降回从库（revert_promote：重新 CHANGE MASTER 指向原主）
|
||
cmd_id = await _send_command(sor, ctx['to_node_id'], 'revert_promote', 'revert_promote')
|
||
ok, _, out = await _wait_command(sor, cmd_id)
|
||
detail.append(f'revert_promote {ctx["to_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}')
|
||
ok_all = ok_all and ok
|
||
elif step_name == 'flip_sync':
|
||
# 再反转一次即恢复
|
||
from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
|
||
nodes = {}
|
||
for nid in (from_id, to_id):
|
||
recs = await sor.R('nodes', {'id': nid})
|
||
if recs:
|
||
nodes[nid] = getattr(recs[0], 'host', '')
|
||
from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '')
|
||
await sor.sqlExe(
|
||
'UPDATE filesync_directories SET source_host=${fh}$, dest_host=${th}$, '
|
||
'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${th}$',
|
||
{'fh': from_host, 'th': to_host, 'now': _now(), 'cid': ctx['cluster_id']})
|
||
detail.append('flip_sync reverted')
|
||
elif step_name == 'converge':
|
||
# 主备换回
|
||
await sor.U('clusters', {
|
||
'id': ctx['cluster_id'],
|
||
'master_node_id': ctx['from_node_id'],
|
||
'standby_node_id': ctx['to_node_id'],
|
||
'status': 'running',
|
||
'updated_at': _now(),
|
||
})
|
||
await sor.U('nodes', {'id': ctx['from_node_id'], 'role': 'master', 'updated_at': _now()})
|
||
await sor.U('nodes', {'id': ctx['to_node_id'], 'role': 'standby', 'updated_at': _now()})
|
||
detail.append('converge reverted')
|
||
# verify / precheck 无需回滚
|
||
return ok_all, '; '.join(detail)
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 编排器主流程（六步状态机）
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def run_switchover(switchover_id):
|
||
"""执行一次切换（在独立 asyncio task 中运行）。"""
|
||
env = ServerEnv()
|
||
async with get_sor_context(env, DBNAME) as sor:
|
||
logs = await sor.R('switchover_logs', {'id': switchover_id})
|
||
if not logs:
|
||
return
|
||
logrec = logs[0]
|
||
ctx = {
|
||
'switchover_id': switchover_id,
|
||
'cluster_id': getattr(logrec, 'cluster_id', ''),
|
||
'from_node_id': getattr(logrec, 'from_node_id', ''),
|
||
'to_node_id': getattr(logrec, 'to_node_id', ''),
|
||
}
|
||
completed = []
|
||
failed_at = None
|
||
fail_detail = ''
|
||
for seq, (step_name, fn) in enumerate(STEPS, start=1):
|
||
await _record_step(sor, switchover_id, seq, step_name, 'running')
|
||
try:
|
||
ok, detail = await fn(sor, ctx)
|
||
except Exception as e:
|
||
ok, detail = False, f'exception: {e}'
|
||
if ok:
|
||
await _record_step(sor, switchover_id, seq, step_name, 'success', output=detail)
|
||
completed.append(step_name)
|
||
else:
|
||
await _record_step(sor, switchover_id, seq, step_name, 'failed', error=detail)
|
||
failed_at = step_name
|
||
fail_detail = detail
|
||
break
|
||
|
||
if failed_at is None:
|
||
# 全部成功
|
||
await sor.U('switchover_logs', {
|
||
'id': switchover_id, 'status': 'completed',
|
||
'details': 'all steps success', 'completed_at': _now(),
|
||
})
|
||
return
|
||
|
||
# 有失败 → 逆向回滚
|
||
rb_ok, rb_detail = await _rollback(sor, ctx, completed)
|
||
final_status = 'completed' if rb_ok else 'rollback_failed'
|
||
# 回滚成功也意味着切换未达成 → 记 failed（原切换失败），但状态收敛
|
||
await sor.U('switchover_logs', {
|
||
'id': switchover_id,
|
||
'status': 'failed' if rb_ok else 'rollback_failed',
|
||
'details': f'failed at {failed_at}: {fail_detail} | rollback: {rb_detail}',
|
||
'completed_at': _now(),
|
||
})
|
||
# 集群恢复（回滚后主备未变）
|
||
await sor.U('clusters', {
|
||
'id': ctx['cluster_id'], 'status': 'running', 'updated_at': _now(),
|
||
})
|
||
# 回滚失败 → 最高级告警冒泡人工
|
||
if not rb_ok:
|
||
await _emit_alert(sor, ctx['cluster_id'], 'critical',
|
||
f'切换回滚失败，需人工介入：{rb_detail}')
|
||
|
||
|
||
async def _emit_alert(sor, cluster_id, severity, message):
|
||
"""写一条最高级告警到 hostwatch_events（松耦合，约定结构）。"""
|
||
try:
|
||
await sor.C('hostwatch_events', {
|
||
'id': getID(),
|
||
'rule_id': '',
|
||
'node_id': '',
|
||
'severity': severity,
|
||
'match_count': 1,
|
||
'sample_message': f'[FOMS cluster={cluster_id}] {message}'[:2000],
|
||
'status': 'open',
|
||
'notified': '0',
|
||
'first_seen_at': _now(),
|
||
'last_seen_at': _now(),
|
||
})
|
||
except Exception:
|
||
pass # 告警失败不阻断主流程
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 对外入口：确认切换 / 健康判定器
|
||
# ═══════════════════════════════════════════
|
||
|
||
async def confirm_switchover(request, params_kw):
|
||
"""人工确认后启动切换（半自动模式的执行入口）。
|
||
|
||
入参：cluster_id（必填），direction（failover/failback，默认 failover）
|
||
行为：创建 switchover_logs(status=in_progress, trigger=manual_confirm)
|
||
+ 置集群 switching + 起异步任务执行六步。
|
||
"""
|
||
env = ServerEnv()
|
||
cluster_id = params_kw.get('cluster_id', '') if hasattr(params_kw, 'get') else ''
|
||
direction = params_kw.get('direction', 'failover') if hasattr(params_kw, 'get') else 'failover'
|
||
if not cluster_id:
|
||
return {'status': 'error', 'message': 'cluster_id required'}
|
||
log_id = getID()
|
||
now = _now()
|
||
async with get_sor_context(env, DBNAME) as sor:
|
||
clusters = await sor.R('clusters', {'id': cluster_id})
|
||
if not clusters:
|
||
return {'status': 'error', 'message': '集群不存在'}
|
||
cluster = clusters[0]
|
||
master_id = getattr(cluster, 'master_node_id', '')
|
||
standby_id = getattr(cluster, 'standby_node_id', '')
|
||
if direction == 'failback':
|
||
from_id, to_id = standby_id, master_id
|
||
else:
|
||
from_id, to_id = master_id, standby_id
|
||
await sor.C('switchover_logs', {
|
||
'id': log_id,
|
||
'cluster_id': cluster_id,
|
||
'event_type': direction,
|
||
'from_node_id': from_id,
|
||
'to_node_id': to_id,
|
||
'trigger': 'manual_confirm',
|
||
'reason': params_kw.get('reason', '人工确认切换') if hasattr(params_kw, 'get') else '人工确认切换',
|
||
'status': 'in_progress',
|
||
'started_at': now,
|
||
'created_at': now,
|
||
})
|
||
await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now})
|
||
# 起异步任务执行（不阻塞 HTTP 响应）
|
||
asyncio.get_event_loop().create_task(run_switchover(log_id))
|
||
return {'status': 'ok', 'message': '切换已启动（异步执行）', 'log_id': log_id}
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 健康判定器（后台循环 + 防抖 + 半自动告警）
|
||
# ═══════════════════════════════════════════
|
||
|
||
# 内存防抖计数：{cluster_id: 连续故障次数}
|
||
_fault_counters = {}
|
||
|
||
|
||
async def health_judge_loop():
|
||
"""后台健康判定：周期扫 active 集群，连续 N 次命中才判故障（防抖）。
|
||
|
||
判定条件（任一）：
|
||
· master 心跳超时（> failover_threshold * check_interval 秒）
|
||
· 复制状态 error/failed
|
||
判故障 → 写 switchover_logs(status=detected, trigger=auto_detect) + 告警。
|
||
半自动：只告警不切，等人 confirm_switchover。
|
||
"""
|
||
env = ServerEnv()
|
||
while True:
|
||
try:
|
||
async with get_sor_context(env, DBNAME) as sor:
|
||
clusters = await sor.sqlExe(
|
||
"SELECT id, master_node_id, check_interval, failover_threshold, status "
|
||
"FROM clusters WHERE status IN ('running','switching')", {})
|
||
for c in (clusters or []):
|
||
cid = getattr(c, 'id', '')
|
||
master_id = getattr(c, 'master_node_id', '')
|
||
interval = int(getattr(c, 'check_interval', 10) or 10)
|
||
threshold = int(getattr(c, 'failover_threshold', 3) or 3)
|
||
if not master_id:
|
||
continue
|
||
# 心跳超时判定
|
||
nodes = await sor.R('nodes', {'id': master_id})
|
||
heartbeat_lost = False
|
||
if nodes:
|
||
hb = _parse_dt(getattr(nodes[0], 'last_heartbeat', None))
|
||
if hb is None or (datetime.now() - hb).total_seconds() > threshold * interval:
|
||
heartbeat_lost = True
|
||
# 复制异常判定（松耦合读 dbbackup 表）
|
||
repl_bad = False
|
||
repls = await sor.sqlExe(
|
||
"SELECT id FROM dbbackup_replications WHERE cluster_id=${cid}$ "
|
||
"AND enabled='1' AND sync_status IN ('error','failed') LIMIT 1",
|
||
{'cid': cid})
|
||
if repls:
|
||
repl_bad = True
|
||
faulty = heartbeat_lost or repl_bad
|
||
if faulty:
|
||
_fault_counters[cid] = _fault_counters.get(cid, 0) + 1
|
||
if _fault_counters[cid] >= threshold:
|
||
# 达到阈值 → 判故障（防抖通过）
|
||
reason = ('心跳超时' if heartbeat_lost else '') + \
|
||
('+' if heartbeat_lost and repl_bad else '') + \
|
||
('复制异常' if repl_bad else '')
|
||
await sor.C('switchover_logs', {
|
||
'id': getID(),
|
||
'cluster_id': cid,
|
||
'event_type': 'fault_detected',
|
||
'from_node_id': master_id,
|
||
'to_node_id': '',
|
||
'trigger': 'auto_detect',
|
||
'reason': reason,
|
||
'status': 'detected',
|
||
'created_at': _now(),
|
||
})
|
||
await _emit_alert(sor, cid, 'critical',
|
||
f'主节点故障判定：{reason}（已连续{threshold}次），请人工确认切换')
|
||
# 重置计数，避免重复告警刷屏（下次再连续 threshold 次才再报）
|
||
_fault_counters[cid] = 0
|
||
else:
|
||
_fault_counters[cid] = 0
|
||
except Exception:
|
||
pass # 判定器自身异常不退出循环
|
||
await asyncio.sleep(JUDGE_INTERVAL)
|
||
|
||
|
||
def register_failover():
|
||
"""注册切换编排器到 ServerEnv + 启动健康判定器后台任务。幂等（防热重载重复起任务）。"""
|
||
global _failover_registered
|
||
env = ServerEnv()
|
||
env.confirm_switchover = confirm_switchover
|
||
if _failover_registered:
|
||
return True
|
||
_failover_registered = True
|
||
# 健康判定器后台任务（由 webapp 的 startup 机制拉起）
|
||
from ahserver.configuredServer import add_startup
|
||
add_startup(health_judge_loop())
|
||
return True
|